dashboard: allow message inspection

2 files changed, 128 insertions, 29 deletions

diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index f31808d..4acc507 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -91,6 +91,16 @@
   #console-box .scroll { flex: 1; }
   .console { font-family: ui-monospace, monospace; font-size: 12px; white-space: pre-wrap; }
   .console .when { color: var(--muted); }
+  /* Room on the left for the bar of a related message, so it covers no text. */
+  .console > div { cursor: pointer; border-left: 3px solid transparent; padding-left: 4px; }
+  /* Of one exchange with the message pointed at or shown (`related`): a lighter grey, and
+     the bar. */
+  .console > div.related { background: rgba(127, 127, 127, .08); border-left-color: var(--self); }
+  .console > div:hover { background: var(--hover, rgba(127, 127, 127, .12)); }
+  .console > div.focused { background: rgba(127, 127, 127, .22); }
+  /* The message takes the box's width and scrolls with it, not in a box of its own. */
+  #stats pre.message { margin: 4px 0 0; font-size: 12px; white-space: pre-wrap;
+    max-height: none; overflow: visible; }
   svg { flex: 1; min-height: 0; width: 100%; display: block; cursor: default; }
   /* Dragged to pan or to move a node, also by touch. */
   /* Dragged, not selected: neither its labels' text nor the picture itself. */
@@ -306,6 +316,11 @@ const pairKey = (a, b) => a < b ? a + " " + b : b + " " + a;
 const isPair = key => key !== null && key.includes(" ");
 function focus(id) {
   focused = id === null || id === state.node.id ? null : id;
+  if (shownMessage !== null) {
+    shownMessage = null;
+    document.querySelectorAll("#console > div.focused").forEach(line => line.classList.remove("focused"));
+    markRelated(null);
+  }
   // In the URL too, so a reload (a new version) keeps it.
   try {
     history.replaceState(null, "", focused === null ? location.pathname
@@ -392,6 +407,26 @@ function renderStats() {
   const title = document.getElementById("figures-title");
   const what = el("span", { class: "what" });
   let shown;
+  const stats = document.getElementById("stats");
+  if (shownMessage !== null) {
+    const message = shownMessage;
+    // Drawn once: a message doesn't change, and drawing it anew would scroll it back up.
+    if (stats.dataset.message === String(message.seq)) return;
+    stats.dataset.message = String(message.seq);
+    what.textContent = message.type + " " + WAYS[message.way] + " " + name(message.peer);
+    const field = (key, value) => el("div", { class: "leaf" },
+      el("span", { class: "key" }, key + ": "), value);
+    title.replaceChildren("Message: ", what);
+    document.getElementById("stats").replaceChildren(
+      field("at", clockTime(new Date(message.at))),
+      field(message.way === "received" ? "from" : "to", message.peer),
+      field("way", message.way),
+      field("type", message.type),
+      field("bytes", String(message.bytes)),
+      el("pre", { class: "message" }, message.text));
+    return;
+  }
+  delete stats.dataset.message;
   if (isPair(focused)) {
     const [a, b] = focused.split(" ");
     what.textContent = name(a) + " – " + name(b);
@@ -1653,8 +1688,8 @@ function dotsFor(kinds, total) {
   }
   return dots;
 }
-// The messages exchanged with each peer since the last update: on the map (`flows`,
-// if shown) and in the console (`logTraffic`).
+// The messages exchanged with each peer since the last update, on the map (`flows`, if
+// shown).
 function trafficArrived() {
   if (!state) return;
   const self = state.node.id, now = performance.now();
@@ -1683,7 +1718,6 @@ function trafficArrived() {
           }
         }
         if (total === 0) continue;
-        logTraffic(way, id, total, kinds);
         if (!dots) continue;
         const shown = dotsFor(kinds, total);
         shown.forEach((dot, i) => {
@@ -1697,43 +1731,76 @@ function trafficArrived() {
   if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows);
 }
 
-// --- Console: what this node exchanged with its peers, a line per peer and way at
-// every update (the same counts as the map's messages), the latest at the bottom,
-// `CONSOLE_LINES` at most. Scrolled along while at the bottom. ---
-const CONSOLE_LINES = 200;
-// A time of day on the 24-hour clock, whatever the browser's locale: 18:04:05.
+// --- Console: the messages this node exchanged with its peers, one by one, as the
+// node logs them (`messages` events, `traffic::MessageLog`), the latest at the bottom,
+// `CONSOLE_LINES` at most. Scrolled along while at the bottom. A click on one shows it
+// in the figures box (`shownMessage`). ---
+const CONSOLE_LINES = 500;
+// A time of day on the 24-hour clock, whatever the browser's locale: 18:04:05.123.
 const clockTime = date => [date.getHours(), date.getMinutes(), date.getSeconds()]
-  .map(part => String(part).padStart(2, "0")).join(":");
+  .map(part => String(part).padStart(2, "0")).join(":") + "." +
+  String(date.getMilliseconds()).padStart(3, "0");
 const size = bytes => bytes < 1024 ? bytes + " B" : bytes < 1048576
   ? (bytes / 1024).toFixed(1) + " KiB" : (bytes / 1048576).toFixed(1) + " MiB";
-// The lines, kept while the box is collapsed and drawn when it's opened (`showConsole`).
-const consoleLines = [];
-// A line: when, what, and the messages by kind, each kind in its colour.
-function consoleLine([when, text, kinds]) {
-  const parts = [];
-  for (const [kind, count] of kinds) {
-    if (parts.length) parts.push(", ");
-    parts.push(el("span", { style: "color: " + kindColour(kind) }, kind), " " + count);
+// The messages, kept while the box is collapsed and drawn when it's opened
+// (`showConsole`).
+const consoleMessages = [];
+// The message shown in the figures box instead of what is in focus; null: none.
+let shownMessage = null;
+const WAYS = { sent: "→", received: "←", refused: "↛" };
+// The exchange marked in the console (a message's `related`): the one of the message
+// pointed at, else of the one shown; null: none.
+let relatedMarked = null;
+function markRelated(key) {
+  relatedMarked = key || null;
+  for (const line of document.getElementById("console").children) {
+    line.classList.toggle("related", relatedMarked !== null && line.dataset.related === relatedMarked);
   }
-  return el("div", {}, el("span", { class: "when" }, when + " "), text, " (", ...parts, ")");
 }
-function logTraffic(way, peer, total, kinds) {
-  const bytes = Object.values(kinds).reduce((sum, grown) => sum + grown.bytes, 0);
-  const line = [clockTime(new Date()),
-    (way === "sent" ? "→ " : "← ") + name(peer) + ": " + total + ", " + size(bytes),
-    Object.entries(kinds).map(([kind, grown]) => [kind, grown.messages])];
-  consoleLines.push(line);
-  if (consoleLines.length > CONSOLE_LINES) consoleLines.shift();
+const shownRelated = () => shownMessage && shownMessage.related;
+// A line: when, which way, with whom, the type (in its kind's colour) and the size.
+function consoleLine(message) {
+  const line = el("div", { "data-seq": String(message.seq),
+    "data-related": message.related || "",
+    class: (shownMessage && shownMessage.seq === message.seq ? "focused" : "") +
+      (relatedMarked !== null && message.related === relatedMarked ? " related" : ""),
+    title: message.way === "refused" ? "refused: the transport wouldn't send it" : "" },
+    el("span", { class: "when" }, clockTime(new Date(message.at)) + " "),
+    WAYS[message.way] + " " + name(message.peer) + " ",
+    el("span", { style: "color: " + kindColour(message.counted) }, message.type),
+    " " + size(message.bytes));
+  // On press, as the peers' rows: lines come and go under the pointer.
+  line.addEventListener("pointerdown", event => {
+    if (event.button !== 0) return;
+    showMessage(shownMessage && shownMessage.seq === message.seq ? null : message);
+  });
+  // The messages of its exchange marked while pointed at; then the shown one's again.
+  line.addEventListener("pointerenter", () => markRelated(message.related));
+  line.addEventListener("pointerleave", () => markRelated(shownRelated()));
+  return line;
+}
+function showMessage(message) {
+  shownMessage = message;
+  const log = document.getElementById("console");
+  for (const line of log.children) {
+    line.classList.toggle("focused", message !== null && line.dataset.seq === String(message.seq));
+  }
+  markRelated(message !== null ? message.related : null);
+  render();
+}
+function messagesArrived(messages) {
+  consoleMessages.push(...messages);
+  consoleMessages.splice(0, Math.max(0, consoleMessages.length - CONSOLE_LINES));
   if (hidden(document.getElementById("console-box"))) return;
   const log = document.getElementById("console");
   const scroll = document.getElementById("console-scroll");
   const following = scroll.scrollTop + scroll.clientHeight >= scroll.scrollHeight - 4;
-  log.append(consoleLine(line));
+  log.append(...messages.slice(-CONSOLE_LINES).map(consoleLine));
   while (log.childElementCount > CONSOLE_LINES) log.firstElementChild.remove();
   if (following) scroll.scrollTop = scroll.scrollHeight;
 }
 function showConsole() {
-  document.getElementById("console").replaceChildren(...consoleLines.map(consoleLine));
+  document.getElementById("console").replaceChildren(...consoleMessages.map(consoleLine));
   const scroll = document.getElementById("console-scroll");
   scroll.scrollTop = scroll.scrollHeight;
 }
@@ -1807,6 +1874,7 @@ function connect() {
     trafficArrived();
   });
   events.addEventListener("document", event => { pulse(); documentChanged(JSON.parse(event.data)); });
+  events.addEventListener("messages", event => messagesArrived(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 732c95b..fdc9b4a 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -72,6 +72,9 @@ struct Hub {
     /// Where the document changed since the last tick, to the clients connected:
     /// `{"paths": [...]}`, or `{"all": true}` if too many or not known.
     documents: broadcast::Sender<Arc<String>>,
+    /// The messages exchanged since the last tick, one by one (`traffic::MessageLog`),
+    /// to the clients connected: a JSON list.
+    messages: broadcast::Sender<Arc<String>>,
     control: Mutex<Control>,
     /// Someone asked for it now (`refresh`): a tick at once, members asked again.
     poke: Notify,
@@ -94,11 +97,13 @@ impl Hub {
         let (latest, _) = watch::channel(None);
         let (patches, _) = broadcast::channel(64);
         let (documents, _) = broadcast::channel(64);
+        let (messages, _) = broadcast::channel(64);
         Arc::new(Self {
             state,
             latest,
             patches,
             documents,
+            messages,
             poke: Notify::new(),
             live: AtomicBool::new(true),
             control: Mutex::new(Control {
@@ -146,6 +151,10 @@ 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.
+        let log = &self.state.traffic.log;
+        log.watch(true);
+        let mut logged = 0u64;
         loop {
             let live = self.live.load(Ordering::Relaxed);
             if Instant::now() >= graph_due {
@@ -182,11 +191,19 @@ impl Hub {
             if let Some(changed) = document_changes(&mut changes) {
                 let _ = self.documents.send(Arc::new(changed.to_string()));
             }
+            let messages = log.after(logged, MESSAGES_AT_MOST);
+            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
             };
             if *self.closing.borrow() {
+                log.watch(false);
                 self.control.lock_or_recover().running = false;
                 return;
             }
@@ -198,6 +215,7 @@ impl Hub {
                 }
                 let mut control = self.control.lock_or_recover();
                 if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until {
+                    log.watch(false);
                     control.running = false;
                     return;
                 }
@@ -210,6 +228,8 @@ impl Hub {
 
 /// Paths changed in a tick listed at most; more: `all`.
 const MAX_CHANGED: usize = 256;
+/// Messages sent to the clients per tick at most, the latest (the console keeps fewer).
+const MESSAGES_AT_MOST: usize = 500;
 
 /// Where the document changed since last asked (`Hub::documents`), if anywhere.
 fn document_changes(
@@ -240,6 +260,7 @@ fn document_changes(
 enum Next {
     Patch(Result<Arc<Published>, broadcast::error::RecvError>),
     Document(Result<Arc<String>, broadcast::error::RecvError>),
+    Messages(Result<Arc<String>, broadcast::error::RecvError>),
     Closing,
 }
 
@@ -507,12 +528,14 @@ async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response {
 /// `GET /admin/dashboard/events`: Server-Sent Events: a `snapshot` first, then a
 /// `patch` whenever something changed (each with its number as the event's ID; a
 /// client that fell behind gets a `snapshot` again), a `document` where the document
-/// changed (`Hub::documents`, at most every tick), and `offline` when the node stops,
-/// which ends the stream.
+/// changed (`Hub::documents`, at most every tick), `messages` with the messages since
+/// the last tick (`Hub::messages`), and `offline` when the node stops, which ends the
+/// stream.
 async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
     // Subscribed before the snapshot is read: no patch after it is missed.
     let mut patches = hub.patches.subscribe();
     let mut documents = hub.documents.subscribe();
+    let mut messages = hub.messages.subscribe();
     hub.want();
     let first = hub.current().await;
     let stream = async_stream::stream! {
@@ -526,6 +549,7 @@ async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
             let next = tokio::select! {
                 received = patches.recv() => Next::Patch(received),
                 changed = documents.recv() => Next::Document(changed),
+                logged = messages.recv() => Next::Messages(logged),
                 () = hub.closed() => Next::Closing,
             };
             let received = match next {
@@ -539,6 +563,13 @@ async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
                     continue;
                 }
                 Next::Document(Err(broadcast::error::RecvError::Closed)) => break,
+                Next::Messages(Ok(logged)) => {
+                    yield Ok(Event::default().event("messages").data(logged.as_str()));
+                    continue;
+                }
+                // Missed ones are only missed in the console.
+                Next::Messages(Err(broadcast::error::RecvError::Lagged(_))) => continue,
+                Next::Messages(Err(broadcast::error::RecvError::Closed)) => break,
                 Next::Closing => {
                     yield Ok(offline_event());
                     break;