From e360bc6da05f1c83c08946763c93e5b16db4e19d Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 12:21:10 +0100 Subject: dashboard: judge peers's clocks --- src/api/dashboard/clocks.rs | 126 +++++++++++++++++++++++++++++++++++++++ src/api/dashboard/dashboard.html | 14 ++++- src/api/dashboard/mod.rs | 43 +++++++++---- 3 files changed, 171 insertions(+), 12 deletions(-) create mode 100644 src/api/dashboard/clocks.rs diff --git a/src/api/dashboard/clocks.rs b/src/api/dashboard/clocks.rs new file mode 100644 index 0000000..313b78e --- /dev/null +++ b/src/api/dashboard/clocks.rs @@ -0,0 +1,126 @@ +//! Every member's clock against the swarm's, for the map: from the clocks each member +//! reports of its peers (`Link::clock`), chained from this node along the narrowest +//! bounds, then taken against their median. A member is off when it differs from the +//! median beyond its chained bound and the limit (`--clock-skew-warn`); with fewer than +//! three members compared, none is (two can't tell whose clock is off). An estimate for +//! the eye: each node judges its own clock by its peers' (`swarm::skew`). + +use std::cmp::Reverse; +use std::collections::{BTreeMap, BinaryHeap, HashMap}; + +use serde::Serialize; + +use crate::transport::Link; + +/// A member's clock against the swarm's. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(super) struct MemberClock { + /// How far it is ahead of the median (behind if negative). + pub(super) offset_ms: i64, + /// Give or take: the bounds chained to it from this node. + pub(super) within_ms: u64, + pub(super) off: bool, +} + +/// From `local` and `reports` (each member's links), with `limit_ms`; by member. +pub(super) fn swarm<'a>( + local: &str, + reports: impl IntoIterator, + limit_ms: u64, +) -> BTreeMap { + // Each way between two members: how far the far end's clock is ahead, give or take. + let mut edges: HashMap<&str, Vec<(&str, i64, u64)>> = HashMap::new(); + for (reporter, links) in reports { + for link in links { + if let Some(clock) = link.clock { + edges.entry(reporter).or_default().push(( + link.peer.as_str(), + clock.offset_ms, + clock.within_ms, + )); + edges.entry(link.peer.as_str()).or_default().push(( + reporter, + -clock.offset_ms, + clock.within_ms, + )); + } + } + } + // Ahead of this node's, along the narrowest chained bounds (Dijkstra). + let mut reached: HashMap<&str, (i64, u64)> = HashMap::new(); + let mut queue = BinaryHeap::from([Reverse((0_u64, local, 0_i64))]); + while let Some(Reverse((within, member, offset))) = queue.pop() { + if reached.contains_key(member) { + continue; + } + reached.insert(member, (offset, within)); + for &(next, step, step_within) in edges.get(member).into_iter().flatten() { + if !reached.contains_key(next) { + queue.push(Reverse(( + within.saturating_add(step_within), + next, + offset.saturating_add(step), + ))); + } + } + } + let mut offsets: Vec = reached.values().map(|(offset, _)| *offset).collect(); + offsets.sort_unstable(); + let median = offsets.get(offsets.len() / 2).copied().unwrap_or(0); + let judged = reached.len() >= 3 && limit_ms > 0; + reached + .into_iter() + .map(|(member, (offset, within))| { + let offset_ms = offset - median; + let off = judged && offset_ms.unsigned_abs() > within.saturating_add(limit_ms); + ( + member.to_owned(), + MemberClock { + offset_ms, + within_ms: within, + off, + }, + ) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::paths::ClockOffset; + + fn link(peer: &str, offset_ms: i64) -> Link { + Link { + peer: peer.into(), + transport: None, + relayed: false, + rtt_us: None, + dormant: false, + others: Vec::new(), + clock: Some(ClockOffset { + offset_ms, + within_ms: 20, + }), + } + } + + /// Chained through the reports: the one clock 10 s ahead is off, wherever it is. + #[test] + fn the_odd_clock_out() { + // a – b – c – d, d's clock 10 s ahead; a only knows b. + let a = [link("b", 0)]; + let b = [link("a", 0), link("c", 5)]; + let c = [link("d", 10_000)]; + let clocks = swarm("a", [("a", &a[..]), ("b", &b[..]), ("c", &c[..])], 2_000); + assert_eq!(clocks.len(), 4); + assert!(clocks["d"].off); + assert!(clocks["d"].offset_ms > 9_900); + assert!(!clocks["a"].off && !clocks["b"].off && !clocks["c"].off); + assert_eq!(clocks["d"].within_ms, 60); + // Two alone: none. + let two = swarm("a", [("a", &[link("b", 10_000)][..])], 2_000); + assert!(two.values().all(|clock| !clock.off)); + } +} diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html index 153d7cb..73b7d30 100644 --- a/src/api/dashboard/dashboard.html +++ b/src/api/dashboard/dashboard.html @@ -90,6 +90,8 @@ .legend .node.silent::before { background: var(--panel); } .legend .node.self::before { border-color: var(--self); background: var(--self); } .legend .node.dormant-node::before { border: 2.5px solid var(--bad); } + .legend .node.clock-off::before { width: 14px; height: 14px; background: none; + border: 1.5px dashed var(--warn); } .legend .node.focus::before { width: 14px; height: 14px; background: var(--self); border: none; opacity: .35; } .scroll { overflow: auto; min-height: 0; } @@ -150,6 +152,7 @@ reports its links reports none (no LINKS_REPLY) dormant (to this node) + clock off the swarm's in focus (click a node)
@@ -649,16 +652,23 @@ function renderGraph() { element.addEventListener("dblclick", () => unpin(id)); hoverable(element, id); }; + // A clock off the swarm's: a dashed orange ring. + if (((state.graph.clocks || {})[id] || {}).off) { + children.push(drawing("clock " + id, "circle", { cx: p.x, cy: p.y, r: 11, fill: "none", + stroke: "var(--warn)", "stroke-width": 1.5, "stroke-dasharray": "3 2" })); + } // Dormant as this node sees it (`/peers`): a red outline. const dormant = (state.peers.peers[id] || {}).status === "dormant"; const circle = drawing("node " + id, "circle", { cx: p.x, cy: p.y, r: member.self ? 8 : 6, fill: member.self ? "var(--self)" : reported ? "var(--fg)" : "var(--panel)", stroke: member.self ? "var(--self)" : dormant ? "var(--bad)" : "var(--fg)", "stroke-width": dormant ? 2.5 : 1.5 }, pressed); - const clock = (state.peers.peers[id] || {}).clock; + // Its clock against the swarm's, as the members' reports chain it (the node's). + const clock = ((state.graph.clocks || {})[id]); tooltip(circle, id + (reported ? "" : " (no links reported)") + (dormant ? " (dormant)" : "") + - (clockOff(clock) ? " (clock " + seconds(clock.offsetMs) + " off this node's)" : "")); + (clock && clock.off ? " (clock " + seconds(clock.offsetMs) + " off the swarm's, ±" + + clock.withinMs + " ms)" : "")); const label = drawing("name " + id, "text", upright({ x: p.x + 10, y: p.y + 4, "font-size": 12, fill: "var(--fg)", cursor: "pointer" }), pressed); text(label, name(id)); diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs index c284f6f..310846a 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -12,7 +12,10 @@ //! The map: the graph the member list makes (`topology::intended_graph`, as this node //! sees it), and the links every member measures: subscribed to with LINKS while anyone //! looks (renewed every `LINKS_EVERY`, ended when nobody does; `swarm::links`), so a -//! member tells of a change as it happens; the same answers for all clients. +//! member tells of a change as it happens; the same answers for all clients. With the +//! clocks the links carry, every member's clock against the swarm's (`clocks`). + +mod clocks; use std::collections::{BTreeMap, BTreeSet}; use std::convert::Infallible; @@ -271,19 +274,33 @@ async fn build(state: &AppState, graph: &Graph) -> JsonValue { .iter() .map(|(member, neighbours)| (member.clone(), json!(neighbours))) .collect(); + let own = links::own_links(state).await; + let replies: Vec<_> = state + .links + .replies() + .into_iter() + .filter(|(_, age, _)| *age <= LINKS_KEPT_FOR) + .collect(); let mut measured = Map::new(); measured.insert( local.to_owned(), - json!({ "secsAgo": 0, "links": links_json(&links::own_links(state).await) }), + json!({ "secsAgo": 0, "links": links_json(&own) }), ); - for (member, age, member_links) in state.links.replies() { - if age <= LINKS_KEPT_FOR { - measured.insert( - member, - json!({ "secsAgo": age.as_secs(), "links": links_json(&member_links) }), - ); - } + for (member, age, member_links) in &replies { + measured.insert( + member.clone(), + json!({ "secsAgo": age.as_secs(), "links": links_json(member_links) }), + ); } + let clocks = clocks::swarm( + local, + std::iter::once((local, &own[..])).chain( + replies + .iter() + .map(|(member, _, member_links)| (member.as_str(), &member_links[..])), + ), + u64::try_from(state.clock_skew_warn.as_millis()).unwrap_or(u64::MAX), + ); // Everyone on the map: in the member list's graph, or measured. let mut members_json = Map::new(); for member in graph.intended.keys().chain(measured.keys()) { @@ -297,7 +314,12 @@ async fn build(state: &AppState, graph: &Graph) -> JsonValue { "page": page().0, "stats": stats, "peers": peers, - "graph": { "members": members_json, "intended": intended, "measured": measured }, + "graph": { + "members": members_json, + "intended": intended, + "measured": measured, + "clocks": clocks, + }, }) } @@ -311,6 +333,7 @@ fn links_json(links: &[Link]) -> JsonValue { "relayed": link.relayed, "rttMs": link.rtt_us.map(|us| f64::from(us) / 1000.0), "dormant": link.dormant, + "clock": link.clock, "others": link.others.iter().map(|other| json!({ "transport": other.transport, "rttMs": other.rtt_us.map(|us| f64::from(us) / 1000.0), -- cgit v1.3.1