dashboard: allow message inspection

1 files changed, 159 insertions, 1 deletions

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