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.

2 files changed, 129 insertions, 105 deletions

diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index 05c233c..b7b45aa 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -2897,12 +2897,11 @@ function unpin(id) {
 graph.addEventListener("mousemove", event => { if (event.target === graph) hover(null); });
 graph.addEventListener("mouseleave", () => hover(null));
 
-// --- Messages on the map: this node's traffic with each peer (`/peers`, `traffic`),
-// as it grew since the last update, and each subscribed node's messages with its peers
-// (`remoteMessagesArrived`), drawn as dots running along the link the way the messages
-// went. Counts come with each update (every 2 s), so the dots are spread
-// over the next 2 s, a few at most per link and way: a picture of the traffic, not a
-// record of it. As dots, all at one speed (`flowSpeed`), so over a longer link they
+// --- Messages on the map: this node's messages with its peers (`ownMessagesArrived`)
+// and each subscribed node's (`remoteMessagesArrived`), as their streams bring them,
+// soon after they went, drawn as dots running along the link the way the messages
+// went, as far apart as they were sent, a few at most per link and way of a batch: a
+// picture of the traffic, not a record of it. As dots, all at one speed (`flowSpeed`), so over a longer link they
 // arrive later; as flashes, each crosses its link as fast as the message did (half the
 // link's round trip) and leaves an afterglow that draws back into where it went
 // (`FLASH_GLOW_MS`). Dots are kept above everything else on the map (`flows`, after
@@ -2980,9 +2979,6 @@ function oneWayMs(from, to) {
   return FLASH_UNKNOWN_MS;
 }
 let lastTraffic = null;
-// When the last update's traffic came, by the wall clock (it runs on while the computer
-// sleeps, unlike `performance.now()` on some systems).
-let lastTrafficAt = 0;
 const flows = [];
 let flowFrame = null;
 // Messages and bytes by kind, of a peer's `traffic.sent` or `.received`.
@@ -3058,17 +3054,12 @@ function dotsFor(kinds, total) {
   }
   return dots;
 }
-// The messages exchanged with each peer since the last update, on the map (`flows`, if
-// shown).
+// This node's traffic with each peer, as the update brings it: its rates, and what its
+// peers should have received (`deliveredHere`). The map runs messages from the message
+// logs instead, as they come (`runMessages`).
 function trafficArrived() {
   if (!state) return;
-  const self = state.node.id, now = performance.now();
-  // Only for an update in its time: not the first after a gap (the computer slept, the
-  // stream stalled), whose counts grew over all of it, nor those queued meanwhile and
-  // arriving at once with it, each of which would add its dots to the same 2 s.
-  const wall = Date.now(), gap = wall - lastTrafficAt;
-  lastTrafficAt = wall;
-  const dots = dotsNow(gap);
+  const self = state.node.id;
   const current = new Map();
   for (const [id, peer] of Object.entries(state.peers.peers)) {
     if (peer.traffic) {
@@ -3076,53 +3067,76 @@ function trafficArrived() {
         received: messages(peer.traffic.received) });
     }
   }
-  // What a subscribed node sent this one runs from its own messages, as it left
-  // (`remoteMessagesArrived`), not from what this one counted received.
-  if (lastTraffic && dots) addFlows(self, current, lastTraffic,
-    (way, id) => way === "received" && logging(id));
   if (lastTraffic) deliveredHere(self, current, lastTraffic);
   measureRates(current);
   lastTraffic = current;
 }
-// Dots for `node`'s traffic with each peer, as it grew from `before` to `current` (by
-// peer: `sent`, `received`, each messages and bytes by kind), spread over the next 2 s;
-// but where `skip(way, peer)` (drawn already from another's).
-function addFlows(node, current, before, skip) {
+// Whether messages run on the map now: shown, and seen.
+function flowsNow() {
+  return flowsShown && !calm.matches && !document.hidden &&
+    !document.getElementById("map").classList.contains("collapsed");
+}
+// Messages as a stream brought them, this node's or a subscribed node's (each named by
+// both ends, `from` and `to`), run along their links as they come: started as far
+// apart as they were sent (`at`, by the clock of the node that logged them, so only
+// against each other), `FLOWS_AT_MOST` at most per link and way of a batch, shared out
+// by kind (`dotsFor`) where more came. Of a batch that spans more than
+// `FLOW_SPREAD_MS` (one after a gap: the computer slept, the stream stalled), only
+// those of its last `FLOW_SPREAD_MS`.
+function runMessages(list) {
+  if (!flowsNow() || list.length === 0) return;
   const now = performance.now();
-  for (const [id, counts] of current) {
-    const was = before.get(id);
-    if (!was) continue;
-    for (const [way, from, to] of [["sent", node, id], ["received", id, node]]) {
-      if (skip(way, id)) continue;
-      const kinds = {};
-      let total = 0;
-      for (const [kind, count] of Object.entries(counts[way])) {
-        const earlier = was[way][kind] || { messages: 0, bytes: 0 };
-        const grown = Math.max(0, count.messages - earlier.messages);
-        if (grown > 0) {
-          kinds[kind] = { messages: grown, bytes: Math.max(0, count.bytes - earlier.bytes) };
-          total += grown;
-        }
+  const latest = Math.max(...list.map(message => message.at));
+  const recent = list.filter(message => latest - message.at <= FLOW_SPREAD_MS);
+  const first = Math.min(...recent.map(message => message.at));
+  const links = new Map();
+  for (const message of recent) {
+    const transport = schemeOf(message.address);
+    const key = message.from + " " + message.to + " " + (transport || "");
+    const link = links.get(key) ||
+      { from: message.from, to: message.to, transport, messages: [] };
+    link.messages.push(message);
+    links.set(key, link);
+  }
+  for (const { from, to, transport, messages: sent } of links.values()) {
+    const start = message => now + (message.at - first);
+    if (sent.length <= FLOWS_AT_MOST) {
+      for (const message of sent) {
+        flows.push({ from, to, transport, start: start(message), kind: message.counted,
+          r: dotRadius(message.bytes) });
       }
-      if (total === 0) continue;
-      const shown = dotsFor(kinds, total);
-      shown.forEach((dot, i) => {
-        flows.push({ from, to, start: now + i * FLOW_SPREAD_MS / shown.length,
-          kind: dot.kind, r: dotRadius(dot.bytes) });
-      });
+      continue;
+    }
+    const kinds = {};
+    for (const message of sent) {
+      const kind = kinds[message.counted] || (kinds[message.counted] = { messages: 0, bytes: 0 });
+      kind.messages += 1;
+      kind.bytes += message.bytes;
     }
+    const span = Math.max(...sent.map(message => message.at)) - Math.min(...sent.map(message => message.at));
+    const from0 = start(sent.reduce((a, b) => (a.at <= b.at ? a : b)));
+    const shown = dotsFor(kinds, sent.length);
+    shown.forEach((dot, i) => {
+      flows.push({ from, to, transport, start: from0 + i * span / shown.length,
+        kind: dot.kind, r: dotRadius(dot.bytes) });
+    });
   }
   if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows);
 }
-// Whether dots are drawn now, for an update whose traffic came `gap` ms after the last:
-// in its time (not the first after a gap, the computer asleep or the stream stalled,
-// whose counts grew over all of it), shown, and seen.
-function dotsNow(gap) {
-  return gap >= FLOW_SPREAD_MS / 4 && gap <= FLOW_SPREAD_MS * 2.5 && flowsShown &&
-    !calm.matches && !document.hidden &&
-    !document.getElementById("map").classList.contains("collapsed");
+// This node's messages, as its stream brings them (`messages` events): listed in the
+// console, and run along its links, those it sent, and those it received from a node
+// whose messages the page doesn't read (a subscribed node's run from it, as they left).
+function ownMessagesArrived(list) {
+  messagesArrived(list);
+  if (!state) return;
+  const self = state.node.id;
+  runMessages(list.filter(message => message.way === "sent" ||
+    (message.way === "received" && !logging(message.peer)))
+    .map(message => ({ ...message,
+      from: message.way === "sent" ? self : message.peer,
+      to: message.way === "sent" ? message.peer : self })));
 }
-// A subscribed node's messages (`follow`), those of a tick: listed in the console and
+// A subscribed node's messages (`follow`), as they come: listed in the console and
 // run along its links, as this node's own. Its messages with this node are this node's
 // to show, and what it received from another subscribed node that one's (as what it
 // sent): each message once. Each named by both ends (`from`, `to`) and the node that
@@ -3130,10 +3144,8 @@ function dotsNow(gap) {
 function remoteMessagesArrived(id, entry, list) {
   if (!state) return;
   const self = state.node.id;
-  const wall = Date.now(), gap = wall - (entry.messagesAt || 0);
-  entry.messagesAt = wall;
-  // A tick's messages come whole, those since the last, up to the latest 500: one
-  // missing between (a busy tick) leaves what was delivered unknown for now.
+  // Its messages come whole, those since the last event, up to the latest 500: one
+  // missing between (a busy moment) leaves what was delivered unknown for now.
   if (list.length > 0) {
     if (entry.lastSeq !== undefined && list[0].seq !== entry.lastSeq + 1) forgetDeliveries(id);
     entry.lastSeq = list[list.length - 1].seq;
@@ -3157,29 +3169,7 @@ function remoteMessagesArrived(id, entry, list) {
     }
   }
   checkDeliveries();
-  if (dotsNow(gap) && running.length > 0) {
-    const now = performance.now();
-    const links = new Map();
-    for (const message of running) {
-      const transport = schemeOf(message.address);
-      const key = message.from + " " + message.to + " " + (transport || "");
-      const link = links.get(key) ||
-        { from: message.from, to: message.to, transport, kinds: {}, total: 0 };
-      const kind = link.kinds[message.counted] || (link.kinds[message.counted] = { messages: 0, bytes: 0 });
-      kind.messages += 1;
-      kind.bytes += message.bytes;
-      link.total += 1;
-      links.set(key, link);
-    }
-    for (const { from, to, transport, kinds, total } of links.values()) {
-      const shown = dotsFor(kinds, total);
-      shown.forEach((dot, i) => {
-        flows.push({ from, to, transport, start: now + i * FLOW_SPREAD_MS / shown.length,
-          kind: dot.kind, r: dotRadius(dot.bytes) });
-      });
-    }
-    if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows);
-  }
+  runMessages(running);
   if (kept.length > 0) messagesArrived(kept);
 }
 // Whether the page reads `id`'s messages one by one: a subscribed node's, while they
@@ -3646,7 +3636,7 @@ function connect() {
     remakesArrived();
   });
   events.addEventListener("document", event => { pulse(); documentChanged(JSON.parse(event.data)); });
-  events.addEventListener("messages", event => messagesArrived(JSON.parse(event.data)));
+  events.addEventListener("messages", event => ownMessagesArrived(JSON.parse(event.data)));
   // The node stops: it says so, and ends the stream. Back once it is.
   events.addEventListener("offline", () => {
     events.close(); live = false;
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| {