diff options
| -rw-r--r-- | src/node.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 4 | ||||
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 7 | ||||
| -rw-r--r-- | src/transport/identity.rs | 5 | ||||
| -rw-r--r-- | src/transport/mod.rs | 14 | ||||
| -rw-r--r-- | src/transport/paths.rs | 19 | ||||
| -rw-r--r-- | src/transport/wire.rs | 39 |
7 files changed, 59 insertions, 30 deletions
diff --git a/src/node.rs b/src/node.rs index fda317c..9f1bd6b 100644 --- a/src/node.rs +++ b/src/node.rs @@ -78,6 +78,7 @@ impl Node { identity, Arc::clone(&traffic), Some(Arc::clone(&firewall)), + Arc::clone(&config.clock), )) as Arc<dyn Transport>, None => config.transport, }; diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index b57feaf..18d29d5 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -33,7 +33,7 @@ pub(crate) async fn handle_message( return Ok(()); } // PING and PONG only say the peer is alive (`keepalive::note_alive`). - if matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) { + if matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong { .. }) { note_alive(state, &peer, &message).await; return Ok(()); } @@ -165,7 +165,7 @@ pub(crate) async fn handle_message( transport::ParsedMessage::Who(_) | transport::ParsedMessage::Me(_) | transport::ParsedMessage::Ping(_) - | transport::ParsedMessage::Pong(_) => {} + | transport::ParsedMessage::Pong { .. } => {} } Ok(()) diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 5e03d0e..11817cc 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -48,12 +48,15 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes let now = Instant::now(); match message { ParsedMessage::Ping(nonce) if !state.transport.answers_pings() => { - let pong = ParsedMessage::Pong(*nonce); + let pong = ParsedMessage::Pong { + nonce: *nonce, + at: state.clock.now_millis(), + }; if let Err(error) = state.transport.send_message(peer, &pong).await { debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed"); } } - ParsedMessage::Pong(_) => { + ParsedMessage::Pong { .. } => { if let Some(rtt) = state.transport.rtt(peer) { state.lock_rtt().measured(peer, rtt, now); } diff --git a/src/transport/identity.rs b/src/transport/identity.rs index 7ce8832..eb7bb93 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -244,12 +244,14 @@ impl Book { impl IdentityTransport { /// Over `transport`, whose addresses have `config.scheme`, and `config.more`. - /// `firewall`: the debug firewall, if any, which paths applies by address. + /// `firewall`: the debug firewall, if any, which paths applies by address; `clock`: + /// the node's wall clock (`Paths::new`). pub(crate) fn new( transport: Arc<dyn Transport>, config: IdentityConfig, traffic: Arc<Traffic>, firewall: Option<Arc<Firewall>>, + clock: Arc<crate::clock::Clock>, ) -> Self { let node_id = config.key.id(); let shared = Arc::new_cyclic(|shared: &std::sync::Weak<Shared>| { @@ -261,6 +263,7 @@ impl IdentityTransport { .collect(), Some(Arc::clone(&traffic)), firewall.is_some().then_some(gate), + clock, )), firewall, key: config.key, diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 903dd07..9a82cbe 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -219,8 +219,12 @@ pub(crate) enum ParsedMessage { Me(IdentityProof), /// Asks for a PONG with this nonce, over the same path (`transport::paths`). Ping(u64), - /// Answers a PING. - Pong(u64), + /// Answers a PING: its nonce, and when it was answered by the answerer's clock + /// (`at`), for comparing clocks (`transport::paths`). + Pong { + nonce: u64, + at: crate::clock::UnixMillis, + }, /// Asks a member how it is linked to its peers, for a map of the swarm /// (`swarm::links`). Links(LinksRequest), @@ -367,7 +371,7 @@ impl ParsedMessage { ParsedMessage::Who(_) => MessageKind::Who, ParsedMessage::Me(_) => MessageKind::Me, ParsedMessage::Ping(_) => MessageKind::Ping, - ParsedMessage::Pong(_) => MessageKind::Pong, + ParsedMessage::Pong { .. } => MessageKind::Pong, ParsedMessage::Links(_) => MessageKind::Links, ParsedMessage::LinksReply(_) => MessageKind::LinksReply, } @@ -389,7 +393,7 @@ impl ParsedMessage { ParsedMessage::Who(_) | ParsedMessage::Me(_) | ParsedMessage::Ping(_) - | ParsedMessage::Pong(_) => None, + | ParsedMessage::Pong { .. } => None, } } } @@ -735,7 +739,7 @@ impl<T: std::fmt::Display> std::fmt::Display for Addressed<T> { /// consumer hands them on as soon as they are read (`hand_on_probe`), not after the /// messages read before them are handled (which can take a while: sync). pub(crate) fn is_probe(message: &ParsedMessage) -> bool { - matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) + matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong { .. }) } /// Hands a PING or PONG (`is_probe`) to `handler` in a task of its own; `scheme` is the diff --git a/src/transport/paths.rs b/src/transport/paths.rs index 2e7ef76..fcfadaa 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -47,6 +47,7 @@ use anyhow::{Context, Result, anyhow}; use tokio::time::Instant; use tracing::{debug, info}; +use crate::clock::Clock; use crate::locks::MutexExt; use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats}; use crate::transport::traffic::Traffic; @@ -394,17 +395,20 @@ pub(crate) struct Paths { traffic: Option<Arc<Traffic>>, /// The debug firewall, if any. gate: Option<Weak<dyn Gate>>, + /// The node's wall clock, which its PONGs say the time by. + clock: Arc<Clock>, /// A std mutex: never held across an await. table: Mutex<Table>, } impl Paths { /// Over `transports`, by scheme, in order of preference; its own PINGs and PONGs - /// counted in `traffic`, the firewall through `gate`. + /// counted in `traffic`, the firewall through `gate`, PONGs stamped by `clock`. pub(crate) fn new( transports: Vec<(&'static str, Arc<dyn Transport>)>, traffic: Option<Arc<Traffic>>, gate: Option<Weak<dyn Gate>>, + clock: Arc<Clock>, ) -> Self { Self { links: transports @@ -413,6 +417,7 @@ impl Paths { .collect(), traffic, gate, + clock, table: Mutex::default(), } } @@ -766,7 +771,10 @@ impl Paths { /// (`traffic`). async fn send_probe(&self, address: &str, pong: bool, nonce: u64) -> Result<()> { let message = if pong { - ParsedMessage::Pong(nonce) + ParsedMessage::Pong { + nonce, + at: self.clock.now_millis(), + } } else { ParsedMessage::Ping(nonce) }; @@ -839,7 +847,7 @@ impl Paths { } }); } - ParsedMessage::Pong(nonce) => { + ParsedMessage::Pong { nonce, .. } => { let location = self.location(&address); let mut table = self.table.lock_or_recover(); table.located(&address, &location); @@ -959,6 +967,7 @@ mod tests { ], None, None, + Arc::default(), )) }; (on_both("a"), on_both("b")) @@ -1252,11 +1261,13 @@ mod tests { vec![("sim", net.transport("b") as Arc<dyn Transport>)], None, Some(Arc::downgrade(&gate)), + Arc::default(), )); let a = Arc::new(Paths::new( vec![("sim", net.transport("a") as Arc<dyn Transport>)], None, None, + Arc::default(), )); let received = |paths: &Arc<Paths>| { let (sender, received) = tokio::sync::mpsc::unbounded_channel(); @@ -1350,11 +1361,13 @@ mod tests { ], None, None, + Arc::default(), ); let b = Arc::new(Paths::new( vec![("sim", net.transport("b") as Arc<dyn Transport>)], None, None, + Arc::default(), )); let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); let handler: Arc<PathHandler> = Arc::new(move |from, _, _| { diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 00bc739..3daba0d 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 = 11; +pub(crate) const VERSION: u8 = 12; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -81,9 +81,8 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> { ParsedMessage::Who(proof) | ParsedMessage::Me(proof) => { rmp_serde::encode::write(&mut payload, proof)? } - ParsedMessage::Ping(nonce) | ParsedMessage::Pong(nonce) => { - rmp_serde::encode::write(&mut payload, nonce)? - } + ParsedMessage::Ping(nonce) => rmp_serde::encode::write(&mut payload, nonce)?, + 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)?, } @@ -117,7 +116,8 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> { } else if is(MessageKind::Ping) { ParsedMessage::Ping(rmp_serde::from_slice(body)?) } else if is(MessageKind::Pong) { - ParsedMessage::Pong(rmp_serde::from_slice(body)?) + let (nonce, at) = rmp_serde::from_slice(body)?; + ParsedMessage::Pong { nonce, at } } else if is(MessageKind::Links) { ParsedMessage::Links(rmp_serde::from_slice(body)?) } else if is(MessageKind::LinksReply) { @@ -419,7 +419,10 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { ParsedMessage::Who(proof.clone()), ParsedMessage::Me(proof), ParsedMessage::Ping(7), - ParsedMessage::Pong(u64::MAX), + ParsedMessage::Pong { + nonce: u64::MAX, + at: crate::clock::UnixMillis(5), + }, ParsedMessage::Links(crate::transport::LinksRequest { epoch, epoch_seq: 1, @@ -615,7 +618,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("0b0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("0c0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -635,7 +638,7 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "0b0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", + "0c0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) @@ -651,7 +654,7 @@ mod tests { }); assert_eq!( hex(&encode(&status).unwrap()), - format!("0b03969290c400{epoch_hex}01c3c2c2") + format!("0c03969290c400{epoch_hex}01c3c2c2") ); let who = ParsedMessage::Who(IdentityProof { @@ -660,7 +663,7 @@ mod tests { }); assert_eq!( hex(&encode(&who).unwrap()), - format!("0b0692c420{}c440{}", "08".repeat(32), "09".repeat(64)) + format!("0c0692c420{}c440{}", "08".repeat(32), "09".repeat(64)) ); let ParsedMessage::Me(proof) = decode( &encode(&ParsedMessage::Me(IdentityProof { @@ -674,14 +677,16 @@ mod tests { }; assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64])); - assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "0b0807"); - assert_eq!( - hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()), - "0b09cfffffffffffffffff" - ); - let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else { + assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "0c0807"); + let pong = ParsedMessage::Pong { + nonce: u64::MAX, + at: crate::clock::UnixMillis(5), + }; + assert_eq!(hex(&encode(&pong).unwrap()), "0c0992cfffffffffffffffff05"); + let ParsedMessage::Pong { nonce, at } = decode(&[VERSION, 9, 0x92, 0x2a, 0x07]).unwrap() + else { panic!("not a PONG"); }; - assert_eq!(nonce, 42); + assert_eq!((nonce, at.as_u64()), (42, 7)); } } |