1 files changed, 25 insertions, 107 deletions

diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index 78e8352..9cb4ea4 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -10,16 +10,14 @@
 //! 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: 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. 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.
+//! sees it), and the links this node measures (`own_links`). Other members' links the
+//! page reads from them itself, from the nodes it subscribes to
+//! (`GET /admin/dashboard/links/own/events`, there).
+
+mod links;
 
 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;
 
@@ -35,23 +33,16 @@ use tokio::time::Instant;
 
 use crate::locks::MutexExt;
 use crate::state::AppState;
-use crate::swarm::{links, members, topology};
-use crate::transport::{Hex, Link};
+use crate::swarm::{members, topology};
+use crate::transport::Hex;
+use links::Link;
 
 /// How often the snapshot is built and compared, while anyone looks.
 const TICK: Duration = Duration::from_secs(2);
-/// How often the subscriptions to the members' links are renewed (each renewal is
-/// answered: a reply at least this often), and the graph recomputed; changes come in
-/// between, as they happen (`swarm::links`).
-const LINKS_EVERY: Duration = Duration::from_secs(30);
-/// How long a subscription lasts unless renewed: a node that is gone stops it.
-const LINKS_LEASE: Duration = Duration::from_secs(90);
-/// A member's links older than this are left out (it didn't answer since).
-const LINKS_KEPT_FOR: Duration = Duration::from_secs(90);
+/// How often the member list's graph is recomputed (it costs about `members²`).
+const GRAPH_EVERY: Duration = Duration::from_secs(30);
 /// 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. A snapshot's text is
 /// made when first asked for (a client connecting, or one that fell behind), not every
@@ -104,11 +95,8 @@ struct Hub {
     /// to the clients connected: a JSON list.
     messages: broadcast::Sender<Arc<String>>,
     control: Mutex<Control>,
-    /// Someone asked for it now (`refresh`): a tick at once, members asked again.
+    /// Someone asked for it now (`refresh`): a tick at once, the graph recomputed.
     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>,
@@ -133,7 +121,6 @@ impl Hub {
             documents,
             messages,
             poke: Notify::new(),
-            live: AtomicBool::new(true),
             control: Mutex::new(Control {
                 running: false,
                 wanted_until: Instant::now(),
@@ -186,12 +173,11 @@ impl Hub {
         log.watch(true);
         let mut logged = 0u64;
         loop {
-            let live = self.live.load(Ordering::Relaxed);
             if Instant::now() >= graph_due {
-                graph.refresh(&self.state, live).await;
-                graph_due = Instant::now() + LINKS_EVERY;
+                graph.refresh(&self.state);
+                graph_due = Instant::now() + GRAPH_EVERY;
             }
-            let snapshot = build(&self.state, &graph, live).await;
+            let snapshot = build(&self.state, &graph).await;
             let stripped = without_ages(&snapshot);
             let changed = last.as_ref().is_none_or(|(_, old)| *old != stripped);
             if changed {
@@ -236,11 +222,7 @@ impl Hub {
                 return;
             }
             if unwanted() {
-                // Nobody looks: the members needn't tell us any more. Unless someone
-                // came meanwhile, then it goes on (and subscribes again).
-                for member in self.state.links.subscribed() {
-                    links::unsubscribe(&self.state, &member).await;
-                }
+                // Nobody looks: it stops. Unless someone came meanwhile, then it goes on.
                 let mut control = self.control.lock_or_recover();
                 if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until {
                     log.watch(false);
@@ -314,34 +296,20 @@ enum Next {
     Closing,
 }
 
-/// The intended graph, recomputed every `LINKS_EVERY` (it costs about `members²`), when
-/// members are asked for their links too: subscribed to, if `live`.
+/// The intended graph, recomputed every `GRAPH_EVERY`.
 #[derive(Default)]
 struct Graph {
     intended: BTreeMap<String, BTreeSet<String>>,
 }
 
 impl Graph {
-    async fn refresh(&mut self, state: &AppState, live: bool) {
-        let lease = if live { LINKS_LEASE } else { Duration::ZERO };
+    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)
-        };
-        let members: Vec<String> = members
-            .into_iter()
-            .filter(|member| member != state.local_id())
-            .take(MAX_ASKED)
-            .collect();
-        // Those we can't reach, through peers that can (`links::ask_all`).
-        links::ask_all(state, members, lease).await;
     }
 }
 
-/// 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 snapshot: the node, its figures, its peers, and the map.
+async fn build(state: &AppState, graph: &Graph) -> 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();
@@ -369,43 +337,18 @@ async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue {
         .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();
-    // With each its epoch: the page shows who entered a new one.
-    let epochs = state.links.epochs();
-    let epoch_json = |epoch: Option<crate::transport::Epoch>| {
-        epoch.map(|epoch| json!(crate::document::replica::EpochRecord::from(epoch)))
-    };
+    // With its epoch: the page shows who entered a new one.
     let mut measured = Map::new();
     measured.insert(
         local.to_owned(),
         json!({ "secsAgo": 0, "links": links_json(&own),
-            "epoch": epoch_json(Some(state.local_epoch())) }),
+            "epoch": crate::document::replica::EpochRecord::from(state.local_epoch()) }),
     );
-    for (member, age, member_links) in &replies {
-        measured.insert(
-            member.clone(),
-            json!({ "secsAgo": age.as_secs(), "links": links_json(member_links),
-                "epoch": epoch_json(epochs.get(member).copied()) }),
-        );
-    }
     // Everyone on the map: in the member list's graph, measured, or at the other end of
-    // a link measured. A node that joined through others is linked to them before its
+    // a link measured. A node that joined through this one is linked to it before its
     // registration reaches this node, and before the graph is computed again
-    // (`LINKS_EVERY`); it shows then, hollow (no links of its own reported).
-    let linked: BTreeSet<&String> = own
-        .iter()
-        .chain(
-            replies
-                .iter()
-                .flat_map(|(_, _, member_links)| member_links.iter()),
-        )
-        .map(|link| &link.peer)
-        .collect();
+    // (`GRAPH_EVERY`); it shows then, hollow (no links of its own reported).
+    let linked: BTreeSet<&String> = own.iter().map(|link| &link.peer).collect();
     let mut members_json = Map::new();
     for member in graph.intended.keys().chain(measured.keys()).chain(linked) {
         members_json.insert(
@@ -423,7 +366,6 @@ async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue {
             "members": members_json,
             "intended": intended,
             "measured": measured,
-            "live": live,
         },
         "remakes": remakes,
     })
@@ -522,7 +464,6 @@ 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))
         .route("/admin/dashboard/tree", get(writer_tree))
         .route("/admin/dashboard/links/own", get(own_links_now))
         .route("/admin/dashboard/links/own/events", get(own_links_events))
@@ -667,7 +608,7 @@ async fn page_html() -> Html<&'static str> {
 }
 
 /// `POST /admin/dashboard/refresh`: a tick now rather than at the next `TICK`, and the
-/// members asked for their links again; what changed comes as events, as ever.
+/// graph recomputed; what changed comes as events, as ever.
 async fn refresh(Extension(hub): Extension<Arc<Hub>>) -> StatusCode {
     hub.want();
     hub.poke.notify_one();
@@ -675,29 +616,6 @@ async fn refresh(Extension(hub): Extension<Arc<Hub>>) -> StatusCode {
 }
 
 #[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
-}
-
-#[derive(serde::Deserialize)]
 struct TreeQuery {
     writer: String,
 }