dashboard: where a stream's lag comes from

A stream's lag in the tooltip of `behind: …` was one figure, from when
a node logged a message to when the page had it: a second and a half
for a node a LAN away said nothing of where it went.

The node now stamps each `messages` event with when it sent it
(`{"sent": …, "messages": […]}`), and the tooltip splits each stream's
lag: how long its batches waited in the node, from logging the first
message to sending the batch (the node's own clock: exact), and how
long they took on the way, from sending to arriving. This node's own
stream is always listed, beside the three slowest.

2 files changed, 63 insertions, 25 deletions

diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index 1d95db1..c99d773 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -2565,7 +2565,11 @@ function follow(id, delay = 2000) {
   const followMessages = (wait = 2000) => {
     if (controller.signal.aborted) return;
     read(messagesUrl, controller.signal, () => { entry.messagesLive = true; wait = 2000; },
-      (kind, data) => { if (kind === "messages") remoteMessagesArrived(id, entry, JSON.parse(data)); })
+      (kind, data) => {
+        if (kind !== "messages") return;
+        const batch = JSON.parse(data);
+        remoteMessagesArrived(id, entry, batch.messages, batch.sent);
+      })
       .catch(() => {
         entry.messagesLive = false;
         if (!controller.signal.aborted) setTimeout(() => followMessages(Math.min(60000, wait * 2)), wait);
@@ -3108,12 +3112,22 @@ function aheadOf(node) {
   const self = clocksKnown[state.node.id], it = clocksKnown[node];
   return self && it ? it.offsetMs - self.offsetMs : 0;
 }
-// `node`'s stream brought a batch whose earliest message lagged `lag` ms (by the
-// browser's clock, against this node's): the most of it, which the delay must cover.
-function noteLag(node, lag) {
-  const now = performance.now();
+// A batch of `node`'s stream came, sent by it at `sent` (its clock): how long its
+// messages lagged, the most of them (by the browser's clock against this node's,
+// `aheadOf`), which the delay must cover; and of that, how long they waited in the node
+// (`sent` against when it logged them: its own clock, exact) and how long the batch took
+// on the way (from `sent`, by the clocks as known). A batch of a gap (lagging more than
+// `REPLAY_MOST_MS`) counts for nothing: it would hold the map that far behind for a while.
+function noteStream(node, list, sent) {
+  if (!state || list.length === 0) return;
+  const ahead = aheadOf(node), wall = Date.now(), now = performance.now();
+  const earliest = Math.min(...list.map(message => message.at));
+  const lag = wall - (earliest - ahead);
+  if (lag > REPLAY_MOST_MS) return;
+  const inNode = typeof sent === "number" ? sent - earliest : null;
+  const way = typeof sent === "number" ? wall - (sent - ahead) : null;
   const kept = (lags.get(node) || []).filter(([at]) => now - at <= REPLAY_WINDOW_MS);
-  kept.push([now, lag]);
+  kept.push([now, lag, inNode, way]);
   lags.set(node, kept);
   for (const [other, samples] of lags) {
     if (!samples.some(([at]) => now - at <= REPLAY_WINDOW_MS)) lags.delete(other);
@@ -3139,12 +3153,17 @@ function showReplay() {
 }
 function explainReplay() {
   const shown = document.getElementById("replay-delay");
-  if (!shown) return;
+  if (!shown || !state) return;
   const s = ms => (ms / 1000).toFixed(2) + " s";
-  const each = [...lags].map(([node, samples]) => ({ node,
-    least: Math.min(...samples.map(([, lag]) => lag)),
-    most: Math.max(...samples.map(([, lag]) => lag)) }))
-    .sort((x, y) => y.most - x.most);
+  const range = values => values.length === 0 ? "?"
+    : Math.min(...values) === Math.max(...values) ? s(values[0])
+    : s(Math.min(...values)) + "–" + s(Math.max(...values));
+  const self = state.node.id;
+  const each = [...lags].map(([node, samples]) => {
+    const of = i => samples.map(sample => sample[i]).filter(value => typeof value === "number");
+    return { node, least: Math.min(...of(1)), most: Math.max(...of(1)),
+      inNode: of(2), way: of(3) };
+  }).sort((x, y) => y.most - x.most);
   if (each.length === 0) {
     shown.title = "Messages run when they went, this far behind real time; no stream yet.";
     return;
@@ -3152,13 +3171,21 @@ function explainReplay() {
   const slowest = each[0];
   const bound = slowest.most + REPLAY_MARGIN_MS < REPLAY_LEAST_MS ? ", raised to the least, " + s(REPLAY_LEAST_MS)
     : slowest.most + REPLAY_MARGIN_MS > REPLAY_MOST_MS ? ", cut to the most, " + s(REPLAY_MOST_MS) : "";
+  const line = ({ node, least, most, inNode, way }) =>
+    (node === self ? "this node (" + name(node) + ")" : name(node)) + ": " + s(most) +
+    " (least " + s(least) + "); in the node " + range(inNode) + ", on the way " + range(way);
+  // The three slowest, and this node's own stream always.
+  const listed = each.slice(0, 3);
+  const own = each.find(stream => stream.node === self);
+  if (own && !listed.includes(own)) listed.push(own);
   shown.title = "Messages run when they went, this far behind real time, so the slowest" +
     " stream's are in too:\n" + s(replayDelay) + " = " + name(slowest.node) + "'s lag, the most" +
     " of any stream in the last 30 s (" + s(slowest.most) + "), + " + s(REPLAY_MARGIN_MS) +
-    bound + "\n\nThe slowest streams (the most and the least each lagged, last 30 s):\n" +
-    each.slice(0, 3).map(({ node, least, most }, i) => (i + 1) + ". " + name(node) + ": " +
-      s(most) + " (least " + s(least) + ")").join("\n") +
-    (each.length > 3 ? "\n(" + (each.length - 3) + " more)" : "");
+    bound + "\n\nThe slowest streams, and this node's (the most and the least each lagged," +
+    " last 30 s; of that, how long a batch waited in its node, from logging its first" +
+    " message to sending it, and took on the way, from sending to arriving):\n" +
+    listed.map((stream, i) => (each.indexOf(stream) + 1) + ". " + line(stream)).join("\n") +
+    (each.length > listed.length ? "\n(" + (each.length - listed.length) + " more)" : "");
 }
 // Messages as `node`'s stream brought them (each named by both ends, `from` and `to`;
 // `arrived`: logged by its receiver, the sender's not read, so it went its link's
@@ -3172,9 +3199,6 @@ function runMessages(list, node) {
   // On the timeline: when the message was logged, as `performance.now()` has it, now
   // and behind.
   const placed = list.map(message => ({ ...message, lag: wall - (message.at - ahead) }));
-  // Not those of a gap, which would hold the map that far behind for a while.
-  const fresh = placed.filter(message => message.lag <= REPLAY_MOST_MS);
-  if (fresh.length > 0) noteLag(node, Math.max(...fresh.map(message => message.lag)));
   if (!flowsNow()) return;
   const links = new Map();
   for (const message of placed) {
@@ -3214,9 +3238,10 @@ function runMessages(list, node) {
 // 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) {
+function ownMessagesArrived(list, sent) {
   messagesArrived(list);
   if (!state) return;
+  noteStream(state.node.id, list, sent);
   const self = state.node.id;
   runMessages(list.filter(message => message.way === "sent" ||
     (message.way === "received" && !logging(message.peer)))
@@ -3229,8 +3254,9 @@ function ownMessagesArrived(list) {
 // 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
 // logged it (`node`), its number and exchange made its own (`node:seq`).
-function remoteMessagesArrived(id, entry, list) {
+function remoteMessagesArrived(id, entry, list, sent) {
   if (!state) return;
+  noteStream(id, list, sent);
   const self = state.node.id;
   // 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.
@@ -3744,7 +3770,10 @@ function connect() {
     remakesArrived();
   });
   events.addEventListener("document", event => { pulse(); documentChanged(JSON.parse(event.data)); });
-  events.addEventListener("messages", event => ownMessagesArrived(JSON.parse(event.data)));
+  events.addEventListener("messages", event => {
+    const batch = JSON.parse(event.data);
+    ownMessagesArrived(batch.messages, batch.sent);
+  });
   // 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 1d63f38..7a643d0 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -250,13 +250,21 @@ impl Hub {
             };
             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));
-            }
+            let batch = json!({ "sent": now_millis(), "messages": messages });
+            let _ = self.messages.send(Arc::new(batch.to_string()));
         }
     }
 }
 
+/// Unix milliseconds by the system's clock, as the message log stamps messages
+/// (`Logged::at`): when a batch of them goes, so the page tells the time they waited
+/// in the node from the time they took on the way.
+fn now_millis() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map_or(0, |since| since.as_millis() as u64)
+}
+
 /// A task stopped when this is dropped.
 struct AbortOnDrop(tokio::task::JoinHandle<()>);
 
@@ -605,7 +613,8 @@ async fn own_messages_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoRe
                     Some(json)
                 })
                 .collect();
-            yield Ok(Event::default().event("messages").data(JsonValue::Array(messages).to_string()));
+            let batch = json!({ "sent": now_millis(), "messages": messages });
+            yield Ok(Event::default().event("messages").data(batch.to_string()));
         }
     };
     (