dashboard: messages as they come

The node sent its messages to the page every 2 s (`messages` events,
this node's and a subscribed node's), and this node's ran on the map
from its counts, which come with the state: the map showed a message 2
to 4 s after it went, the messages of each 2 s spread evenly over the
next.

The message log now tells those waiting of each message logged
(MessageLog::logged_after), and both streams send what was logged a
tenth of a second after the first of it (MESSAGES_GATHERED): a busy
node about ten events a second, a quiet one each message soon after it
went. The page runs messages as their streams bring them, this node's
from its own log too (with the addresses its PINGs went to), as far
apart as they were sent; more than six on a link and way at once are
shared out by kind over the moment, and after a gap only the last 2 s
run. The counts still give the rates and what was delivered.

1 files changed, 57 insertions, 23 deletions

diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index d6485d7..1d63f38 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -37,6 +37,11 @@ use crate::swarm::{members, topology};
 use crate::transport::Hex;
 use links::Link;
 
+/// How long messages logged are gathered before they are sent on, after the first: a
+/// busy node sends about ten events a second at most, a quiet one each message soon
+/// after it went.
+const MESSAGES_GATHERED: Duration = Duration::from_millis(100);
+
 /// How often the snapshot is built and compared, while anyone looks.
 const TICK: Duration = Duration::from_secs(2);
 /// How often the member list's graph is recomputed (it costs about `members²`).
@@ -91,8 +96,8 @@ struct Hub {
     /// `{"paths": [...]}`, or `{"all": true}` if too many or not known. Shared by every page
     /// open; each is told of the paths its user may read at least partly (`changed_for`).
     documents: broadcast::Sender<Arc<String>>,
-    /// The messages exchanged since the last tick, one by one (`traffic::MessageLog`),
-    /// to the clients connected: a JSON list.
+    /// The messages exchanged, one by one (`traffic::MessageLog`), to the clients
+    /// connected as they come (`forward_messages`): a JSON list.
     messages: broadcast::Sender<Arc<String>>,
     control: Mutex<Control>,
     /// Someone asked for it now (`refresh`): a tick at once, the graph recomputed.
@@ -168,10 +173,11 @@ impl Hub {
         let mut graph = Graph::default();
         let mut graph_due = Instant::now();
         let mut changes = self.state.change_events.subscribe();
-        // The messages one by one, from now on, while it runs.
+        // The messages one by one, from now on, while it runs: sent on as they come.
         let log = &self.state.traffic.log;
         log.watch(true);
-        let mut logged = 0u64;
+        let forwarding = tokio::spawn(Arc::clone(&self).forward_messages());
+        let _forwarding = AbortOnDrop(forwarding);
         loop {
             if Instant::now() >= graph_due {
                 graph.refresh(&self.state);
@@ -204,14 +210,6 @@ impl Hub {
             if let Some(changed) = document_changes(&mut changes) {
                 let _ = self.documents.send(Arc::new(changed.to_string()));
             }
-            let mut messages = log.after(logged, MESSAGES_AT_MOST);
-            by_node(&self.state, &mut messages).await;
-            if let Some(last) = messages.last() {
-                logged = last.seq;
-                if let Ok(json) = serde_json::to_string(&messages) {
-                    let _ = self.messages.send(Arc::new(json));
-                }
-            }
             let unwanted = || {
                 self.patches.receiver_count() == 0
                     && Instant::now() >= self.control.lock_or_recover().wanted_until
@@ -236,6 +234,38 @@ impl Hub {
     }
 }
 
+impl Hub {
+    /// The messages logged, to every client connected (`messages`), as they come:
+    /// gathered for `MESSAGES_GATHERED` after the first, the latest `MESSAGES_AT_MOST`.
+    /// For as long as the producer runs.
+    async fn forward_messages(self: Arc<Self>) {
+        let log = &self.state.traffic.log;
+        let mut logged = log.latest();
+        loop {
+            log.logged_after(logged).await;
+            tokio::time::sleep(MESSAGES_GATHERED).await;
+            let mut messages = log.after(logged, MESSAGES_AT_MOST);
+            let Some(last) = messages.last() else {
+                continue;
+            };
+            logged = last.seq;
+            by_node(&self.state, &mut messages).await;
+            if let Ok(json) = serde_json::to_string(&messages) {
+                let _ = self.messages.send(Arc::new(json));
+            }
+        }
+    }
+}
+
+/// A task stopped when this is dropped.
+struct AbortOnDrop(tokio::task::JoinHandle<()>);
+
+impl Drop for AbortOnDrop {
+    fn drop(&mut self) {
+        self.0.abort();
+    }
+}
+
 /// Paths changed in a tick listed at most; more: `all`.
 /// The messages the paths layer logged under an address (its own PINGs and PONGs) under
 /// the node there instead, where one is known: the address it proved itself at, or is
@@ -536,9 +566,9 @@ impl Drop for Watching {
 }
 
 /// `GET /admin/dashboard/messages/own/events`: server-sent events, `messages` with the
-/// messages this node sent and received since the last (every `TICK`, those of the
-/// tick, none too, as the page's own `messages` events have them, without each message
-/// written out), for a page of another member subscribed to this node: its console lists them,
+/// messages this node sent and received since the last, as they come (gathered for
+/// `MESSAGES_GATHERED` after the first), as the page's own `messages` events have them,
+/// without each message written out, for a page of another member subscribed to this node: its console lists them,
 /// and its map runs them along this node's links. `offline` when the node stops, which
 /// ends it. The message log is kept while it is open.
 async fn own_messages_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
@@ -548,21 +578,25 @@ async fn own_messages_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoRe
     let mut logged = log.latest();
     let stream = async_stream::stream! {
         let _watching = watching;
-        let mut every = tokio::time::interval(TICK);
         loop {
+            // As they come, gathered a moment (`MESSAGES_GATHERED`).
+            let log = &hub.state.traffic.log;
             tokio::select! {
-                _ = every.tick() => {}
+                () = async {
+                    log.logged_after(logged).await;
+                    tokio::time::sleep(MESSAGES_GATHERED).await;
+                } => {}
                 () = hub.closed() => {
                     yield Ok::<_, Infallible>(offline_event());
                     break;
                 }
             }
-            // Every tick, none too: the page times its dots by them.
-            let mut messages = hub.state.traffic.log.after(logged, MESSAGES_AT_MOST);
-            if let Some(last) = messages.last() {
-                logged = last.seq;
-                by_node(&hub.state, &mut messages).await;
-            }
+            let mut messages = log.after(logged, MESSAGES_AT_MOST);
+            let Some(last) = messages.last() else {
+                continue;
+            };
+            logged = last.seq;
+            by_node(&hub.state, &mut messages).await;
             let messages: Vec<JsonValue> = messages
                 .into_iter()
                 .filter_map(|message| {