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.

3 files changed, 70 insertions, 29 deletions

diff --git a/docs/dashboard.md b/docs/dashboard.md
index 0d1500b..28f31a1 100644
--- a/docs/dashboard.md
+++ b/docs/dashboard.md
@@ -147,9 +147,12 @@ the time its node logged it, taken back by how far that node's clock is ahead of
 node's (as the map chains the clocks), a while behind real time. How far behind
 (`behind: …` after the sliders) is the most any stream lagged within the last 30 s and
 a tenth of a second more; its tooltip shows how it came about (which node's stream
-lagged the most, by how much, and the margin) and the three slowest streams, each with
-the most and the least it lagged, so a slow path shows. Lags are measured by the
-browser's clock against this node's: one that is off adds to all of them alike. Messages of different nodes then run as far apart as they went,
+lagged the most, by how much, and the margin) and the three slowest streams and this
+node's own, each with the most and the least it lagged, and of that how long its
+batches waited in their node (from logging the first message to sending the batch: the
+node's own clock, exact) and took on the way (from sending to arriving), so a slow node
+or a slow path shows. Lags are measured by the browser's clock against this node's:
+one that is off adds to all of them alike. Messages of different nodes then run as far apart as they went,
 give or take how well their clocks are known (within the bounds the clock chain has). A
 message the page knows only from its receiver (its sender's messages aren't read: not
 subscribed) arrives at the time its receiver logged it, starting a crossing before.
@@ -484,7 +487,7 @@ members compared, none is.
 **Messages**, on the map and in the console, come from a log the node keeps only while
 someone looks (`traffic::MessageLog`, the latest 1000), sent as a `messages` event as
 they come: a tenth of a second after the first of those since the last, the latest 500
-of them. The per-peer message counts in the state (`/peers`, `traffic`) give the rates
+of them, with when the node sent the event (`{"sent": …, "messages": […]}`). The per-peer message counts in the state (`/peers`, `traffic`) give the rates
 and what was delivered.
 
 **The document** is read through the store API, a level at a time as it is opened
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()));
         }
     };
     (