diff options
| -rw-r--r-- | src/swarm/links.rs | 27 | ||||
| -rw-r--r-- | src/transport/mod.rs | 3 | ||||
| -rw-r--r-- | src/transport/paths.rs | 2 | ||||
| -rw-r--r-- | src/transport/wire.rs | 18 |
4 files changed, 39 insertions, 11 deletions
diff --git a/src/swarm/links.rs b/src/swarm/links.rs index 82cd991..d5238f3 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -25,7 +25,7 @@ use tracing::debug; use crate::locks::MutexExt; use crate::state::AppState; use crate::swarm::{members, topology}; -use crate::transport::paths; +use crate::transport::paths::{self, ClockOffset}; use crate::transport::{ Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, }; @@ -128,6 +128,7 @@ pub(crate) async fn own_links(state: &AppState) -> Vec<Link> { .rtt(&peer) .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)), dormant, + clock: state.transport.clock_offset(&peer), peer, }); } @@ -354,9 +355,16 @@ pub(crate) async fn watch(state: AppState) { } /// Whether `new` says something `old` didn't: another peer, transport, relay or -/// dormancy, or a round trip measured or moved by more than a fifth (and 1 ms). Round -/// trips wobble; small moves aren't news. +/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock +/// measured or moved beyond both bounds. Round trips and clock samples wobble; small +/// moves aren't news. fn changed(old: &[Link], new: &[Link]) -> bool { + let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) { + (Some(a), Some(b)) => { + a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms) + } + (a, b) => a.is_some() != b.is_some(), + }; let moved = |a: Option<u32>, b: Option<u32>| match (a, b) { (Some(a), Some(b)) => { let by = a.abs_diff(b); @@ -371,6 +379,7 @@ fn changed(old: &[Link], new: &[Link]) -> bool { || a.relayed != b.relayed || a.dormant != b.dormant || moved(a.rtt_us, b.rtt_us) + || clock_moved(a.clock, b.clock) || a.others.len() != b.others.len() || a.others.iter().zip(&b.others).any(|(x, y)| { x.transport != y.transport @@ -392,6 +401,7 @@ mod tests { rtt_us, dormant: false, others: Vec::new(), + clock: None, } } @@ -408,5 +418,16 @@ mod tests { assert!(changed(&old, &[dormant])); // Small round trips: a millisecond's wobble isn't news. assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))])); + // A clock: news when measured, and when it moved beyond both bounds. + let clocked = |offset_ms| Link { + clock: Some(ClockOffset { + offset_ms, + within_ms: 50, + }), + ..link("a", Some(10_000)) + }; + assert!(changed(&old, &[clocked(0)])); + assert!(!changed(&[clocked(0)], &[clocked(80)])); + assert!(changed(&[clocked(0)], &[clocked(5_000)])); } } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 1ce4cd1..048fe31 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -442,6 +442,9 @@ pub(crate) struct Link { /// The peer's other transports (`swarm::links::MAX_OTHERS` at most): where else /// it proved itself, those paths up. pub(crate) others: Vec<OtherPath>, + /// How far the peer's clock is ahead of the member's (`paths::ClockOffset`), if + /// measured: with every member's, a map of the swarm shows whose clock is off. + pub(crate) clock: Option<paths::ClockOffset>, } /// A transport of a link besides the one in use, in a LINKS_REPLY. diff --git a/src/transport/paths.rs b/src/transport/paths.rs index 2ef3a12..7ebde70 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -114,7 +114,7 @@ const CLOCK_KEPT: Duration = Duration::from_secs(300); /// PONG: `offset_ms`, give or take `within_ms`. The PONG says its sender's time when it /// answered, which was after our PING went and before the PONG came: within half the /// round trip of its middle, however differently long the two ways took. -#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct ClockOffset { pub(crate) offset_ms: i64, diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 3daba0d..46cea4c 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 = 12; +pub(crate) const VERSION: u8 = 13; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -444,6 +444,10 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { rtt_us: None, relayed: false, }], + clock: Some(crate::transport::paths::ClockOffset { + offset_ms: -2_500, + within_ms: 40, + }), }], }), ] @@ -618,7 +622,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("0c0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("0d0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -638,7 +642,7 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "0c0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", + "0d0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) @@ -654,7 +658,7 @@ mod tests { }); assert_eq!( hex(&encode(&status).unwrap()), - format!("0c03969290c400{epoch_hex}01c3c2c2") + format!("0d03969290c400{epoch_hex}01c3c2c2") ); let who = ParsedMessage::Who(IdentityProof { @@ -663,7 +667,7 @@ mod tests { }); assert_eq!( hex(&encode(&who).unwrap()), - format!("0c0692c420{}c440{}", "08".repeat(32), "09".repeat(64)) + format!("0d0692c420{}c440{}", "08".repeat(32), "09".repeat(64)) ); let ParsedMessage::Me(proof) = decode( &encode(&ParsedMessage::Me(IdentityProof { @@ -677,12 +681,12 @@ mod tests { }; assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64])); - assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "0c0807"); + assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "0d0807"); let pong = ParsedMessage::Pong { nonce: u64::MAX, at: crate::clock::UnixMillis(5), }; - assert_eq!(hex(&encode(&pong).unwrap()), "0c0992cfffffffffffffffff05"); + assert_eq!(hex(&encode(&pong).unwrap()), "0d0992cfffffffffffffffff05"); let ParsedMessage::Pong { nonce, at } = decode(&[VERSION, 9, 0x92, 0x2a, 0x07]).unwrap() else { panic!("not a PONG"); |