dashboard: allow delayed links

1 files changed, 46 insertions, 9 deletions

diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index 0e7b442..732c95b 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -12,16 +12,21 @@
 //! 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. With the
-//! clocks the links carry, every member's clock against the swarm's (`clocks`).
+//! member tells of a change as it happens; the same answers for all clients. Or, if
+//! asked (`POST /admin/dashboard/links?live=false`, for every client, until asked
+//! otherwise or the node restarts), only asked every `LINKS_EVERY`, without
+//! subscribing: fewer messages in the swarm, a map up to `LINKS_EVERY` behind. 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;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, OnceLock};
 use std::time::Duration;
 
+use axum::extract::Query;
 use axum::http::{HeaderValue, StatusCode, header::CONTENT_TYPE};
 use axum::response::sse::{Event, KeepAlive, Sse};
 use axum::response::{Html, IntoResponse, Response};
@@ -70,6 +75,9 @@ struct Hub {
     control: Mutex<Control>,
     /// Someone asked for it now (`refresh`): a tick at once, members asked again.
     poke: Notify,
+    /// The members' links subscribed to (true, at first), or only asked for every
+    /// `LINKS_EVERY` (`set_links`).
+    live: AtomicBool,
     /// True once the node stops: every client gets an `offline` event, and its stream
     /// ends (so the HTTP server needn't wait for clients to go).
     closing: watch::Receiver<bool>,
@@ -92,6 +100,7 @@ impl Hub {
             patches,
             documents,
             poke: Notify::new(),
+            live: AtomicBool::new(true),
             control: Mutex::new(Control {
                 running: false,
                 wanted_until: Instant::now(),
@@ -138,11 +147,12 @@ impl Hub {
         let mut graph_due = Instant::now();
         let mut changes = self.state.change_events.subscribe();
         loop {
+            let live = self.live.load(Ordering::Relaxed);
             if Instant::now() >= graph_due {
-                graph.refresh(&self.state).await;
+                graph.refresh(&self.state, live).await;
                 graph_due = Instant::now() + LINKS_EVERY;
             }
-            let snapshot = build(&self.state, &graph).await;
+            let snapshot = build(&self.state, &graph, live).await;
             let changed = last
                 .as_ref()
                 .is_none_or(|old| without_ages(old) != without_ages(&snapshot));
@@ -234,14 +244,15 @@ enum Next {
 }
 
 /// The intended graph, recomputed every `LINKS_EVERY` (it costs about `members²`), when
-/// members are asked for their links too.
+/// members are asked for their links too: subscribed to, if `live`.
 #[derive(Default)]
 struct Graph {
     intended: BTreeMap<String, BTreeSet<String>>,
 }
 
 impl Graph {
-    async fn refresh(&mut self, state: &AppState) {
+    async fn refresh(&mut self, state: &AppState, live: bool) {
+        let lease = if live { LINKS_LEASE } else { Duration::ZERO };
         self.intended = topology::intended_graph(state).unwrap_or_default();
         let members: Vec<String> = {
             let replica = state.doc.read().await;
@@ -253,13 +264,14 @@ impl Graph {
             .take(MAX_ASKED)
         {
             let state = state.clone();
-            tokio::spawn(async move { links::ask(&state, &member, LINKS_LEASE).await });
+            tokio::spawn(async move { links::ask(&state, &member, lease).await });
         }
     }
 }
 
-/// The snapshot: the node, its figures, its peers, and the map.
-async fn build(state: &AppState, graph: &Graph) -> JsonValue {
+/// The snapshot: the node, its figures, its peers, and the map (`live`: its links
+/// subscribed to).
+async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue {
     // The peers once, for their list and the figures' totals over them.
     let list = super::peers::peer_list(state).await;
     let stats = serde_json::to_value(super::stats::collect(state, &list).await).unwrap_or_default();
@@ -319,6 +331,7 @@ async fn build(state: &AppState, graph: &Graph) -> JsonValue {
             "intended": intended,
             "measured": measured,
             "clocks": clocks,
+            "live": live,
         },
     })
 }
@@ -416,6 +429,7 @@ pub(crate) fn router(state: AppState) -> Router<AppState> {
         .route("/admin/dashboard/state", get(current_state))
         .route("/admin/dashboard/events", get(events))
         .route("/admin/dashboard/refresh", post(refresh))
+        .route("/admin/dashboard/links", post(set_links))
         .layer(Extension(Hub::new(state, closing)))
 }
 
@@ -454,6 +468,29 @@ async fn refresh(Extension(hub): Extension<Arc<Hub>>) -> StatusCode {
     StatusCode::NO_CONTENT
 }
 
+#[derive(serde::Deserialize)]
+struct LinksQuery {
+    live: bool,
+}
+
+/// `POST /admin/dashboard/links?live=…`: the members' links subscribed to (`true`), or
+/// only asked for every `LINKS_EVERY` (`false`), for every client; the subscriptions
+/// ended at once, and the members asked again now.
+async fn set_links(
+    Extension(hub): Extension<Arc<Hub>>,
+    Query(query): Query<LinksQuery>,
+) -> StatusCode {
+    hub.live.store(query.live, Ordering::Relaxed);
+    if !query.live {
+        for member in hub.state.links.subscribed() {
+            links::unsubscribe(&hub.state, &member).await;
+        }
+    }
+    hub.want();
+    hub.poke.notify_one();
+    StatusCode::NO_CONTENT
+}
+
 /// `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();