dashboard: judge peers's clocks

3 files changed, 171 insertions, 12 deletions

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<Item = (&'a str, &'a [Link])>,
+    limit_ms: u64,
+) -> BTreeMap<String, MemberClock> {
+    // 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<i64> = 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 @@
         <span class="node">reports its links</span>
         <span class="node silent">reports none (no LINKS_REPLY)</span>
         <span class="node dormant-node">dormant (to this node)</span>
+        <span class="node clock-off">clock off the swarm's</span>
         <span class="node focus">in focus (click a node)</span>
       </div>
       <div class="legend" id="legend"></div>
@@ -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),