1 files changed, 243 insertions, 1 deletions

diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index b237db2..b7907fb 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;
@@ -29,6 +30,11 @@ impl Counter {
         self.messages += 1;
         self.bytes += bytes as u64;
     }
+
+    fn absorb(&mut self, other: Counter) {
+        self.messages += other.messages;
+        self.bytes += other.bytes;
+    }
 }
 
 #[derive(Debug, Default, Clone, Copy, Serialize)]
@@ -45,9 +51,42 @@ pub(crate) struct ByKind {
     pub(crate) identity: Counter,
     /// PING and PONG (`transport::paths`), counted by the paths layer under the address.
     pub(crate) paths: Counter,
+    /// LINKS and LINKS_REPLY (`swarm::links`).
+    pub(crate) links: Counter,
 }
 
 impl ByKind {
+    fn absorb(&mut self, other: &ByKind) {
+        for kind in [
+            MessageKind::Hello,
+            MessageKind::HelloReply,
+            MessageKind::Sync,
+            MessageKind::Status,
+            MessageKind::StatusReply,
+            MessageKind::Blob,
+            MessageKind::Who,
+            MessageKind::Ping,
+            MessageKind::Links,
+        ] {
+            let theirs = *other.get(kind);
+            self.get_mut(kind).absorb(theirs);
+        }
+    }
+
+    fn get(&self, kind: MessageKind) -> &Counter {
+        match kind {
+            MessageKind::Hello => &self.hello,
+            MessageKind::HelloReply => &self.hello_reply,
+            MessageKind::Sync => &self.sync,
+            MessageKind::Status => &self.status,
+            MessageKind::StatusReply => &self.status_reply,
+            MessageKind::Blob => &self.blob,
+            MessageKind::Who | MessageKind::Me => &self.identity,
+            MessageKind::Ping | MessageKind::Pong => &self.paths,
+            MessageKind::Links | MessageKind::LinksReply => &self.links,
+        }
+    }
+
     fn get_mut(&mut self, kind: MessageKind) -> &mut Counter {
         match kind {
             MessageKind::Hello => &mut self.hello,
@@ -58,6 +97,7 @@ impl ByKind {
             MessageKind::Blob => &mut self.blob,
             MessageKind::Who | MessageKind::Me => &mut self.identity,
             MessageKind::Ping | MessageKind::Pong => &mut self.paths,
+            MessageKind::Links | MessageKind::LinksReply => &mut self.links,
         }
     }
 }
@@ -82,6 +122,169 @@ 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>,
+    /// The address it went to or came from, where the paths layer logged it under that
+    /// (its own PINGs and PONGs) and the node there is known: `peer` is the node then
+    /// (`api::dashboard`).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) address: Option<String>,
+}
+
+/// What ties `message`, with `peer`, to other messages: a PING and its PONG (the
+/// nonce alone: a PING may go to an address and its PONG come back up from the node
+/// there, `paths`), 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 {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();
+        }
+    }
+
+    pub(crate) 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),
+            address: None,
+        });
+        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 {
@@ -115,6 +318,22 @@ impl Traffic {
         self.lock().peers.get(peer).copied()
     }
 
+    /// A node's traffic: counted under it, and under `addresses`, where it is (the
+    /// layers below the node count what they send and receive on their own there: WHO
+    /// and ME, the paths layer's PINGs and PONGs).
+    pub(crate) fn node(&self, node: &str, addresses: &[String]) -> PeerTraffic {
+        let table = self.lock();
+        let mut traffic = table.peers.get(node).copied().unwrap_or_default();
+        for address in addresses.iter().filter(|address| address.as_str() != node) {
+            if let Some(there) = table.peers.get(address) {
+                traffic.sent.absorb(&there.sent);
+                traffic.received.absorb(&there.received);
+                traffic.refused.absorb(&there.refused);
+            }
+        }
+        traffic
+    }
+
     pub(crate) fn total(&self) -> PeerTraffic {
         self.lock().total
     }
@@ -152,6 +371,10 @@ impl Transport for CountingTransport {
         self.inner.delivery_limit()
     }
 
+    fn sync_reply_timeout(&self) -> std::time::Duration {
+        self.inner.sync_reply_timeout()
+    }
+
     fn local_addresses(&self) -> Vec<String> {
         self.inner.local_addresses()
     }
@@ -176,6 +399,14 @@ impl Transport for CountingTransport {
         self.inner.discovered_seeds()
     }
 
+    fn on_seeds_found(&self, found: std::sync::Arc<tokio::sync::Notify>) {
+        self.inner.on_seeds_found(found)
+    }
+
+    fn on_reachable(&self, reachable: std::sync::Arc<tokio::sync::Notify>) {
+        self.inner.on_reachable(reachable)
+    }
+
     fn transports(&self) -> Vec<String> {
         self.inner.transports()
     }
@@ -204,6 +435,10 @@ impl Transport for CountingTransport {
         self.inner.rtt(peer)
     }
 
+    fn clock_offset(&self, peer: &str) -> Option<crate::transport::paths::ClockOffset> {
+        self.inner.clock_offset(peer)
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }
@@ -221,6 +456,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
     }
 
@@ -229,6 +470,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