dashboard: allow message inspection

4 files changed, 292 insertions, 31 deletions

diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 798ef8b..bb8323b 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -604,7 +604,11 @@ paths:
         every 2 s. Each event's ID is its number. A client that falls behind gets a
         `snapshot` again. A `document` event says where the document changed since
         the last (at most every 2 s): `{"paths": [[segment, ...], ...]}`, or
-        `{"all": true}` when too many changed or the node lost count. When the node
+        `{"all": true}` when too many changed or the node lost count. A `messages`
+        event lists the messages the node exchanged since the last (at most every
+        2 s, 500 at most): `seq`, `at` (Unix ms), `way` (`sent`, `received`,
+        `refused`), `peer`, `type` (`SYNC`, …), `counted` (its traffic kind), `bytes`
+        and `text` (the message written out). When the node
         stops, an `offline` event ends the stream (a client reconnects once the node
         is back). Nothing changed, nothing is sent but keep-alive comments.
         Ages (`…SecsAgo`, `secsAgo`, `uptimeSecs`, `dormantForSecs`,
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;
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 4f56d21..e6d4a0c 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -8,7 +8,8 @@
 //! A message dropped by the debug firewall (in the paths layer, below) still counts as
 //! sent (the firewall plays a lossy network) but not as received.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard};
 
 use anyhow::Result;
@@ -85,6 +86,156 @@ struct Table {
 #[derive(Debug, Default)]
 pub(crate) struct Traffic {
     table: Mutex<Table>,
+    /// The messages one by one, while the dashboard looks.
+    pub(crate) log: MessageLog,
+}
+
+/// Messages kept in the log (`MessageLog`) at most, and the text of one.
+const LOGGED: usize = 1000;
+const LOGGED_TEXT: usize = 8 * 1024;
+
+/// The node's messages one by one, for the dashboard's console (`api::dashboard`):
+/// kept only while it looks (`watch`), the latest `LOGGED`. Those this layer sees: the
+/// node's own, not the paths layer's PINGs and PONGs nor the identity layer's WHO and
+/// ME (counted under their addresses, `ByKind`), but the PINGs and PONGs the node is
+/// handed.
+#[derive(Debug, Default)]
+pub(crate) struct MessageLog {
+    watching: AtomicBool,
+    inner: Mutex<LogInner>,
+}
+
+#[derive(Debug, Default)]
+struct LogInner {
+    next: u64,
+    entries: VecDeque<Logged>,
+}
+
+/// One message, as the dashboard shows it.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Logged {
+    /// Counts up, from 1.
+    pub(crate) seq: u64,
+    /// Unix milliseconds, by the system's clock.
+    pub(crate) at: u64,
+    /// `sent`, `received` or `refused` (the transport wouldn't send it).
+    pub(crate) way: &'static str,
+    pub(crate) peer: String,
+    /// The message's type, as the docs name it: `HELLO_REPLY`, `SYNC`, …
+    #[serde(rename = "type")]
+    pub(crate) kind: &'static str,
+    /// Its kind among the counters (`ByKind`'s names): `helloReply`, `sync`, …
+    pub(crate) counted: &'static str,
+    pub(crate) bytes: usize,
+    /// The message, written out (`Debug`; sizes for its bytes), `LOGGED_TEXT` at most.
+    pub(crate) text: String,
+    /// What ties it to other messages (`related`): those with the same are one
+    /// exchange.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) related: Option<String>,
+}
+
+/// What ties `message`, with `peer`, to other messages: a PING and its PONG (the
+/// nonce), a LINKS and its replies (the nonce), a HELLO and its replies (the run of the
+/// node that sent the HELLO, which the reply echoes), a blob's requests and answers, from
+/// any peer (its ID), and a delta offered in a SYNC with the blob it is fetched as. None
+/// for the rest (a STATUS and its reply carry nothing of the kind).
+fn related(peer: &str, message: &ParsedMessage) -> Option<String> {
+    use crate::transport::{Hex, SyncBody};
+    Some(match message {
+        ParsedMessage::Ping(nonce) | ParsedMessage::Pong { nonce, .. } => {
+            format!("ping {peer} {nonce}")
+        }
+        ParsedMessage::Links(request) => format!("links {peer} {}", request.nonce),
+        ParsedMessage::LinksReply(reply) => format!("links {peer} {}", reply.nonce),
+        ParsedMessage::Hello(hello) => format!("hello {peer} {}", Hex(&hello.run_id)),
+        ParsedMessage::HelloReply(reply) => format!("hello {peer} {}", Hex(&reply.run_id)),
+        ParsedMessage::Blob(blob) => format!("blob {}", blob.id),
+        ParsedMessage::Sync(envelope) => match &envelope.body {
+            SyncBody::Delta(spec) => {
+                format!("blob {}", crate::transport::BlobId::Delta(spec.id()))
+            }
+            SyncBody::Message(_) => return None,
+        },
+        _ => return None,
+    })
+}
+
+// Read by the dashboard only.
+#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+impl MessageLog {
+    /// Keeps messages from now on (`true`), or no more, forgetting those kept.
+    pub(crate) fn watch(&self, on: bool) {
+        self.watching.store(on, Ordering::Relaxed);
+        if !on {
+            self.inner.lock_or_recover().entries.clear();
+        }
+    }
+
+    fn note(&self, way: &'static str, peer: &str, message: &ParsedMessage, bytes: usize) {
+        if !self.watching.load(Ordering::Relaxed) {
+            return;
+        }
+        let at = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map_or(0, |since| since.as_millis() as u64);
+        let mut text = format!("{message:#?}");
+        if text.len() > LOGGED_TEXT {
+            let mut cut = LOGGED_TEXT;
+            while !text.is_char_boundary(cut) {
+                cut -= 1;
+            }
+            text.truncate(cut);
+            text.push_str("\n…");
+        }
+        let (kind, counted) = names(message.kind());
+        let mut inner = self.inner.lock_or_recover();
+        inner.next += 1;
+        let seq = inner.next;
+        inner.entries.push_back(Logged {
+            seq,
+            at,
+            way,
+            peer: peer.to_owned(),
+            kind,
+            counted,
+            bytes,
+            text,
+            related: related(peer, message),
+        });
+        if inner.entries.len() > LOGGED {
+            inner.entries.pop_front();
+        }
+    }
+
+    /// The messages after `seq`, the latest `most` of them.
+    pub(crate) fn after(&self, seq: u64, most: usize) -> Vec<Logged> {
+        let inner = self.inner.lock_or_recover();
+        let newer: Vec<&Logged> = inner.entries.iter().filter(|e| e.seq > seq).collect();
+        newer[newer.len().saturating_sub(most)..]
+            .iter()
+            .map(|entry| (*entry).clone())
+            .collect()
+    }
+}
+
+/// A message kind's type name, and its counter's (`ByKind`).
+fn names(kind: MessageKind) -> (&'static str, &'static str) {
+    match kind {
+        MessageKind::Hello => ("HELLO", "hello"),
+        MessageKind::HelloReply => ("HELLO_REPLY", "helloReply"),
+        MessageKind::Sync => ("SYNC", "sync"),
+        MessageKind::Status => ("STATUS", "status"),
+        MessageKind::StatusReply => ("STATUS_REPLY", "statusReply"),
+        MessageKind::Blob => ("BLOB", "blob"),
+        MessageKind::Who => ("WHO", "identity"),
+        MessageKind::Me => ("ME", "identity"),
+        MessageKind::Ping => ("PING", "paths"),
+        MessageKind::Pong => ("PONG", "paths"),
+        MessageKind::Links => ("LINKS", "links"),
+        MessageKind::LinksReply => ("LINKS_REPLY", "links"),
+    }
 }
 
 impl Traffic {
@@ -228,6 +379,12 @@ impl Transport for CountingTransport {
             self.traffic
                 .refused(&message.peer, message.kind, message.payload_len);
         }
+        if let Some(parsed) = &message.message {
+            let way = if result.is_ok() { "sent" } else { "refused" };
+            self.traffic
+                .log
+                .note(way, &message.peer, parsed, message.payload_len);
+        }
         result
     }
 
@@ -236,6 +393,7 @@ impl Transport for CountingTransport {
         self.inner
             .start_consumer(Box::new(move |sender, message, payload_len| {
                 traffic.received(&sender, message.kind(), payload_len);
+                traffic.log.note("received", &sender, &message, payload_len);
                 handler(sender, message, payload_len)
             }))
             .await