1 files changed, 394 insertions, 0 deletions
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
new file mode 100644
index 0000000..adaed10
--- /dev/null
+++ b/src/api/dashboard/mod.rs
@@ -0,0 +1,394 @@
+//! The dashboard (`--features dashboard`): a page showing the node's figures
+//! (`/admin/stats`), its peers (`/peers`) and a map of the swarm, kept up to date
+//! through Server-Sent Events.
+//!
+//! One producer per node serves every client: while at least one is connected (or
+//! asked for the state within `LINGER`), it builds the snapshot every `TICK` and, if
+//! something changed, publishes it, and a JSON merge patch (RFC 7386) from the one
+//! before to every client connected. Nothing changed, nothing is sent (but SSE's
+//! keep-alive comments). Ages (`…SecsAgo`, `uptimeSecs` and the like, `GROWING`) grow by
+//! themselves: they alone changing is no change, and the page counts them on.
+//!
+//! The map: the graph the member list makes (`topology::intended_graph`, as this node
+//! sees it), and the links every member measures, asked for with LINKS every
+//! `LINKS_EVERY` while anyone looks (`swarm::links`), the same answers for all clients.
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::convert::Infallible;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use axum::http::{HeaderValue, header::CONTENT_TYPE};
+use axum::response::sse::{Event, KeepAlive, Sse};
+use axum::response::{Html, IntoResponse, Response};
+use axum::routing::get;
+use axum::{Extension, Router};
+use serde_json::{Map, Value as JsonValue, json};
+use tokio::sync::{broadcast, watch};
+use tokio::time::Instant;
+
+use crate::locks::MutexExt;
+use crate::state::AppState;
+use crate::swarm::{links, members, topology};
+use crate::transport::Link;
+
+/// How often the snapshot is built and compared, while anyone looks.
+const TICK: Duration = Duration::from_secs(2);
+/// How often members are asked for their links, and the graph recomputed.
+const LINKS_EVERY: Duration = Duration::from_secs(30);
+/// A member's links older than this are left out (it didn't answer since).
+const LINKS_KEPT_FOR: Duration = Duration::from_secs(90);
+/// The producer runs on this long after the last request for the state.
+const LINGER: Duration = Duration::from_secs(60);
+/// Members asked for their links at most, each round.
+const MAX_ASKED: usize = 512;
+
+/// A snapshot or a patch as published: its number, and its JSON.
+struct Published {
+    seq: u64,
+    json: String,
+}
+
+/// The producer and what it published, shared by all clients.
+struct Hub {
+    state: AppState,
+    /// The latest snapshot; None while the producer hasn't made one (since it started).
+    latest: watch::Sender<Option<Arc<Published>>>,
+    /// Each patch, to the clients connected.
+    patches: broadcast::Sender<Arc<Published>>,
+    control: Mutex<Control>,
+}
+
+struct Control {
+    running: bool,
+    /// The producer stops once no client is connected and this is past.
+    wanted_until: Instant,
+}
+
+impl Hub {
+    fn new(state: AppState) -> Arc<Self> {
+        let (latest, _) = watch::channel(None);
+        let (patches, _) = broadcast::channel(64);
+        Arc::new(Self {
+            state,
+            latest,
+            patches,
+            control: Mutex::new(Control {
+                running: false,
+                wanted_until: Instant::now(),
+            }),
+        })
+    }
+
+    /// Someone looks: the producer runs (started if it wasn't), for `LINGER` at least.
+    fn want(self: &Arc<Self>) {
+        let mut control = self.control.lock_or_recover();
+        control.wanted_until = Instant::now() + LINGER;
+        if !control.running {
+            control.running = true;
+            // What it published before it stopped is stale.
+            self.latest.send_replace(None);
+            tokio::spawn(Arc::clone(self).run());
+        }
+    }
+
+    /// The latest snapshot, once there is one.
+    async fn current(&self) -> Arc<Published> {
+        let mut latest = self.latest.subscribe();
+        loop {
+            // The borrow (a lock guard, not Send) ends here, before any later await.
+            let published = latest
+                .wait_for(Option::is_some)
+                .await
+                .ok()
+                .and_then(|current| current.clone());
+            if let Some(published) = published {
+                return published;
+            }
+            // The sender lives as long as the hub: not reached.
+            tokio::time::sleep(TICK).await;
+        }
+    }
+
+    async fn run(self: Arc<Self>) {
+        let mut last: Option<JsonValue> = None;
+        let mut seq = 0u64;
+        let mut graph = Graph::default();
+        let mut graph_due = Instant::now();
+        loop {
+            if Instant::now() >= graph_due {
+                graph.refresh(&self.state).await;
+                graph_due = Instant::now() + LINKS_EVERY;
+            }
+            let snapshot = build(&self.state, &graph).await;
+            let changed = last
+                .as_ref()
+                .is_none_or(|old| without_ages(old) != without_ages(&snapshot));
+            if changed {
+                seq += 1;
+                if let Some(old) = &last
+                    && let Some(patch) = merge_patch(old, &snapshot)
+                {
+                    let _ = self.patches.send(Arc::new(Published {
+                        seq,
+                        json: patch.to_string(),
+                    }));
+                }
+                self.latest.send_replace(Some(Arc::new(Published {
+                    seq,
+                    json: snapshot.to_string(),
+                })));
+                last = Some(snapshot);
+            }
+            tokio::time::sleep(TICK).await;
+            {
+                let mut control = self.control.lock_or_recover();
+                if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until {
+                    control.running = false;
+                    return;
+                }
+            }
+        }
+    }
+}
+
+/// The intended graph, recomputed every `LINKS_EVERY` (it costs about `members²`), when
+/// members are asked for their links too.
+#[derive(Default)]
+struct Graph {
+    intended: BTreeMap<String, BTreeSet<String>>,
+}
+
+impl Graph {
+    async fn refresh(&mut self, state: &AppState) {
+        self.intended = topology::intended_graph(state).unwrap_or_default();
+        let members: Vec<String> = {
+            let replica = state.doc.read().await;
+            members::members(&replica.doc)
+        };
+        for member in members
+            .into_iter()
+            .filter(|member| member != state.local_id())
+            .take(MAX_ASKED)
+        {
+            let state = state.clone();
+            tokio::spawn(async move { links::ask(&state, &member).await });
+        }
+    }
+}
+
+/// The snapshot: the node, its figures, its peers, and the map.
+async fn build(state: &AppState, graph: &Graph) -> JsonValue {
+    let stats = serde_json::to_value(super::stats::collect(state).await).unwrap_or_default();
+    let peers = serde_json::to_value(super::peers::peer_list(state).await).unwrap_or_default();
+    let aliases = {
+        let replica = state.doc.read().await;
+        members::aliases(&replica.doc)
+    };
+    let local = state.local_id();
+    let intended: Map<String, JsonValue> = graph
+        .intended
+        .iter()
+        .map(|(member, neighbours)| (member.clone(), json!(neighbours)))
+        .collect();
+    let mut measured = Map::new();
+    measured.insert(
+        local.to_owned(),
+        json!({ "secsAgo": 0, "links": links_json(&links::own_links(state).await) }),
+    );
+    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) }),
+            );
+        }
+    }
+    // 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()) {
+        members_json.insert(
+            member.clone(),
+            json!({ "alias": aliases.get(member), "self": member == local }),
+        );
+    }
+    json!({
+        "node": { "id": local, "alias": state.alias },
+        "stats": stats,
+        "peers": peers,
+        "graph": { "members": members_json, "intended": intended, "measured": measured },
+    })
+}
+
+fn links_json(links: &[Link]) -> JsonValue {
+    links
+        .iter()
+        .map(|link| {
+            json!({
+                "peer": link.peer,
+                "transport": link.transport,
+                "relayed": link.relayed,
+                "rttMs": link.rtt_us.map(|us| f64::from(us) / 1000.0),
+                "dormant": link.dormant,
+            })
+        })
+        .collect()
+}
+
+/// Fields that grow by themselves besides those named `…SecsAgo` (the page knows
+/// them too).
+const GROWING: &[&str] = &[
+    "uptimeSecs",
+    "dormantForSecs",
+    "observedSecs",
+    "lastFromNeighbourSecs",
+    "oldestQuarantineSecs",
+];
+
+/// Whether the field `key` is an age, which grows by itself.
+fn is_age(key: &str) -> bool {
+    key == "secsAgo" || key.ends_with("SecsAgo") || GROWING.contains(&key)
+}
+
+/// `value` without its ages, to tell whether anything else changed.
+fn without_ages(value: &JsonValue) -> JsonValue {
+    match value {
+        JsonValue::Object(object) => JsonValue::Object(
+            object
+                .iter()
+                .filter(|(key, _)| !is_age(key))
+                .map(|(key, value)| (key.clone(), without_ages(value)))
+                .collect(),
+        ),
+        JsonValue::Array(items) => JsonValue::Array(items.iter().map(without_ages).collect()),
+        other => other.clone(),
+    }
+}
+
+/// The JSON merge patch (RFC 7386) from `old` to `new`, if they differ: objects field
+/// by field (a field gone is null), anything else replaced whole. A null in `new` reads
+/// as the field gone, as a merge patch can't tell them apart.
+fn merge_patch(old: &JsonValue, new: &JsonValue) -> Option<JsonValue> {
+    match (old, new) {
+        (JsonValue::Object(old), JsonValue::Object(new)) => {
+            let mut patch = Map::new();
+            for key in old.keys() {
+                if !new.contains_key(key) {
+                    patch.insert(key.clone(), JsonValue::Null);
+                }
+            }
+            for (key, value) in new {
+                match old.get(key) {
+                    Some(previous) => {
+                        if let Some(change) = merge_patch(previous, value) {
+                            patch.insert(key.clone(), change);
+                        }
+                    }
+                    None => {
+                        patch.insert(key.clone(), value.clone());
+                    }
+                }
+            }
+            (!patch.is_empty()).then_some(JsonValue::Object(patch))
+        }
+        (old, new) if old == new => None,
+        (_, new) => Some(new.clone()),
+    }
+}
+
+/// The dashboard's routes (authenticated, as the rest of `/admin`), with their hub.
+pub(crate) fn router(state: AppState) -> Router<AppState> {
+    Router::new()
+        .route("/admin/dashboard", get(page))
+        .route("/admin/dashboard/state", get(current_state))
+        .route("/admin/dashboard/events", get(events))
+        .layer(Extension(Hub::new(state)))
+}
+
+/// `GET /admin/dashboard`: the page.
+async fn page() -> Html<&'static str> {
+    Html(include_str!("dashboard.html"))
+}
+
+/// `GET /admin/dashboard/state`: the latest snapshot, for a client that wants it whole.
+async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response {
+    hub.want();
+    let current = hub.current().await;
+    (
+        [(CONTENT_TYPE, HeaderValue::from_static("application/json"))],
+        current.json.clone(),
+    )
+        .into_response()
+}
+
+/// `GET /admin/dashboard/events`: Server-Sent Events: a `snapshot` first, then a
+/// `patch` whenever something changed (each with its number as the event's ID; a
+/// client that fell behind gets a `snapshot` again).
+async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
+    // Subscribed before the snapshot is read: no patch after it is missed.
+    let mut patches = hub.patches.subscribe();
+    hub.want();
+    let first = hub.current().await;
+    let stream = async_stream::stream! {
+        let mut seen = first.seq;
+        yield Ok::<_, Infallible>(snapshot_event(&first));
+        loop {
+            match patches.recv().await {
+                Ok(patch) if patch.seq <= seen => {}
+                Ok(patch) => {
+                    seen = patch.seq;
+                    yield Ok(Event::default()
+                        .event("patch")
+                        .id(patch.seq.to_string())
+                        .data(patch.json.clone()));
+                }
+                Err(broadcast::error::RecvError::Lagged(_)) => {
+                    let current = hub.current().await;
+                    seen = current.seq;
+                    yield Ok(snapshot_event(&current));
+                }
+                Err(broadcast::error::RecvError::Closed) => break,
+            }
+        }
+    };
+    Sse::new(stream).keep_alive(KeepAlive::default())
+}
+
+fn snapshot_event(published: &Published) -> Event {
+    Event::default()
+        .event("snapshot")
+        .id(published.seq.to_string())
+        .data(published.json.clone())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn merge_patches() {
+        let old = json!({ "a": 1, "b": { "c": 2, "d": [1, 2] }, "gone": true });
+        let new = json!({ "a": 1, "b": { "c": 3, "d": [1, 2] }, "added": "x" });
+        assert_eq!(
+            merge_patch(&old, &new),
+            Some(json!({ "b": { "c": 3 }, "gone": null, "added": "x" }))
+        );
+        assert_eq!(merge_patch(&old, &old), None);
+        assert_eq!(
+            merge_patch(&json!([1]), &json!([1, 2])),
+            Some(json!([1, 2]))
+        );
+    }
+
+    #[test]
+    fn ages_alone_are_no_change() {
+        let old =
+            json!({ "uptimeSecs": 1, "peers": { "p": { "lastSeenSecsAgo": 3, "rttMs": 2 } } });
+        let older =
+            json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 2 } } });
+        assert_eq!(without_ages(&old), without_ages(&older));
+        let moved =
+            json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 5 } } });
+        assert_ne!(without_ages(&old), without_ages(&moved));
+    }
+}