diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/mod.rs | 50 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 3 | ||||
| -rw-r--r-- | src/transport/wire.rs | 41 |
3 files changed, 86 insertions, 8 deletions
diff --git a/src/transport/mod.rs b/src/transport/mod.rs index fd3651b..a15a1f7 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -218,6 +218,11 @@ pub(crate) enum ParsedMessage { Ping(u64), /// Answers a PING. Pong(u64), + /// Asks a member how it is linked to its peers, for a map of the swarm + /// (`swarm::links`). + Links(LinksRequest), + /// Answers a LINKS. + LinksReply(LinksReply), } /// A node's proof that it holds the key of its node ID (`identity`), sending from its @@ -343,6 +348,8 @@ pub(crate) enum MessageKind { Me, Ping, Pong, + Links, + LinksReply, } impl ParsedMessage { @@ -358,6 +365,8 @@ impl ParsedMessage { ParsedMessage::Me(_) => MessageKind::Me, ParsedMessage::Ping(_) => MessageKind::Ping, ParsedMessage::Pong(_) => MessageKind::Pong, + ParsedMessage::Links(_) => MessageKind::Links, + ParsedMessage::LinksReply(_) => MessageKind::LinksReply, } } @@ -372,6 +381,8 @@ impl ParsedMessage { Some((status.epoch, status.epoch_seq)) } 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::Who(_) | ParsedMessage::Me(_) | ParsedMessage::Ping(_) @@ -380,6 +391,45 @@ impl ParsedMessage { } } +/// LINKS: what a member is asked for a map of the swarm (`swarm::links`). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct LinksRequest { + #[serde(with = "crate::transport::wire::codec::epoch")] + pub(crate) epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub(crate) epoch_seq: u32, + /// Repeated in the reply: which request it answers. + pub(crate) nonce: u64, +} + +/// LINKS_REPLY: how the sender is linked to its neighbours, as it measures them. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct LinksReply { + #[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, + /// At most `swarm::links::MAX_LINKS`. + pub(crate) links: Vec<Link>, +} + +/// One of a member's links, in a LINKS_REPLY. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct Link { + /// The peer's node ID. + pub(crate) peer: String, + /// The transport its messages go over now (the scheme of the path in use), if any. + pub(crate) transport: Option<String>, + /// Only through a relay (libp2p). + pub(crate) relayed: bool, + /// The round trip on that path, in microseconds, if measured. + pub(crate) rtt_us: Option<u32>, + /// Silent for long (`peers.md`, dormancy). + pub(crate) dormant: bool, +} + #[derive(Clone, Serialize, Deserialize)] pub(crate) struct HelloRequest { /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index b237db2..f8ca586 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -45,6 +45,8 @@ pub(crate) struct ByKind { pub(crate) identity: Counter, /// PING and PONG (`transport::paths`), counted by the paths layer under the address. pub(crate) paths: Counter, + /// LINKS and LINKS_REPLY (`swarm::links`). + pub(crate) links: Counter, } impl ByKind { @@ -58,6 +60,7 @@ impl ByKind { MessageKind::Blob => &mut self.blob, MessageKind::Who | MessageKind::Me => &mut self.identity, MessageKind::Ping | MessageKind::Pong => &mut self.paths, + MessageKind::Links | MessageKind::LinksReply => &mut self.links, } } } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 53fd460..3e1ab10 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{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 = 6; +pub(crate) const VERSION: u8 = 7; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -29,6 +29,8 @@ fn tag(kind: MessageKind) -> u8 { MessageKind::Me => 7, MessageKind::Ping => 8, MessageKind::Pong => 9, + MessageKind::Links => 10, + MessageKind::LinksReply => 11, } } @@ -67,6 +69,8 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> { ParsedMessage::Ping(nonce) | ParsedMessage::Pong(nonce) => { rmp_serde::encode::write(&mut payload, nonce)? } + ParsedMessage::Links(request) => rmp_serde::encode::write(&mut payload, request)?, + ParsedMessage::LinksReply(reply) => rmp_serde::encode::write(&mut payload, reply)?, } Ok(payload) } @@ -99,6 +103,10 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> { ParsedMessage::Ping(rmp_serde::from_slice(body)?) } else if is(MessageKind::Pong) { ParsedMessage::Pong(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Links) { + ParsedMessage::Links(rmp_serde::from_slice(body)?) + } else if is(MessageKind::LinksReply) { + ParsedMessage::LinksReply(rmp_serde::from_slice(body)?) } else { return Err(WireError::UnknownKind(kind)); }) @@ -396,6 +404,23 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { ParsedMessage::Me(proof), ParsedMessage::Ping(7), ParsedMessage::Pong(u64::MAX), + ParsedMessage::Links(crate::transport::LinksRequest { + epoch, + epoch_seq: 1, + nonce: 3, + }), + ParsedMessage::LinksReply(crate::transport::LinksReply { + epoch, + epoch_seq: 1, + nonce: 3, + links: vec![crate::transport::Link { + peer: "p".into(), + transport: Some("libp2p".into()), + relayed: true, + rtt_us: Some(1500), + dormant: false, + }], + }), ] } @@ -413,7 +438,7 @@ mod tests { #[test] fn every_kind_round_trips() { let samples = samples(); - assert_eq!(samples.len(), 10, "a kind is missing from `samples`"); + assert_eq!(samples.len(), 12, "a kind is missing from `samples`"); for message in samples { let bytes = encode(&message).unwrap(); let decoded = decode(&bytes).unwrap(); @@ -566,7 +591,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("060097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("070097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -586,7 +611,7 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "060199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", + "070199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) @@ -601,7 +626,7 @@ mod tests { }); assert_eq!( hex(&encode(&status).unwrap()), - format!("0603959290c400{epoch_hex}01c3c2") + format!("0703959290c400{epoch_hex}01c3c2") ); let who = ParsedMessage::Who(IdentityProof { @@ -610,7 +635,7 @@ mod tests { }); assert_eq!( hex(&encode(&who).unwrap()), - format!("060692c420{}c440{}", "08".repeat(32), "09".repeat(64)) + format!("070692c420{}c440{}", "08".repeat(32), "09".repeat(64)) ); let ParsedMessage::Me(proof) = decode( &encode(&ParsedMessage::Me(IdentityProof { @@ -624,10 +649,10 @@ mod tests { }; assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64])); - assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "060807"); + assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "070807"); assert_eq!( hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()), - "0609cfffffffffffffffff" + "0709cfffffffffffffffff" ); let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else { panic!("not a PONG"); |