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.

1 files changed, 13 insertions, 4 deletions

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()));
         }
     };
     (