cli: Add `rad node debug` command

radicle-cli/src/commands/node.rs

@@ -33,6 +33,7 @@ Usage
     rad node start [--foreground] [--verbose] [<option>...] [-- <node-option>...]
     rad node stop [<option>...]
     rad node logs [-n <lines>]
+    rad node debug [<option>...]
     rad node connect <nid>@<addr> [<option>...]
     rad node routing [--rid <rid>] [--nid <nid>] [--json] [<option>...]
     rad node events [--timeout <secs>] [-n <count>] [<option>...]
@@ -98,6 +99,7 @@ pub enum Operation {
         lines: usize,
     },
     Status,
+    Debug,
     Sessions,
     Stop,
 }
@@ -113,6 +115,7 @@ pub enum OperationName {
     Start,
     #[default]
     Status,
+    Debug,
     Sessions,
     Stop,
 }
@@ -152,6 +155,7 @@ impl Args for Options {
                     "status" => op = Some(OperationName::Status),
                     "stop" => op = Some(OperationName::Stop),
                     "sessions" => op = Some(OperationName::Sessions),
+                    "debug" => op = Some(OperationName::Debug),
 
                     unknown => anyhow::bail!("unknown operation '{}'", unknown),
                 },
@@ -222,6 +226,7 @@ impl Args for Options {
                 path: path.unwrap_or(PathBuf::from("radicle-node")),
             },
             OperationName::Status => Operation::Status,
+            OperationName::Debug => Operation::Debug,
             OperationName::Sessions => Operation::Sessions,
             OperationName::Stop => Operation::Stop,
         };
@@ -250,6 +255,9 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> {
         Operation::Db { args } => {
             commands::db(&profile, args)?;
         }
+        Operation::Debug => {
+            control::debug(&mut node)?;
+        }
         Operation::Sessions => {
             let sessions = control::sessions(&node)?;
             if let Some(table) = sessions {

radicle-cli/src/commands/node/control.rs

@@ -120,6 +120,13 @@ pub fn stop(node: Node) -> anyhow::Result<()> {
     Ok(())
 }
 
+pub fn debug(node: &mut Node) -> anyhow::Result<()> {
+    let json = node.debug()?;
+    term::json::to_pretty(&json, Path::new("debug.json"))?.print();
+
+    Ok(())
+}
+
 pub fn logs(lines: usize, follow: Option<time::Duration>, profile: &Profile) -> anyhow::Result<()> {
     let logs = profile.home.node().join("node.log");
 

radicle-node/src/control.rs

@@ -199,6 +199,11 @@ where
             }
             Err(e) => return Err(CommandError::Runtime(e)),
         },
+        Command::Debug => {
+            let debug = handle.debug()?;
+
+            CommandResult::Okay(debug).to_writer(writer)?;
+        }
         Command::Shutdown => {
             log::debug!(target: "control", "Shutdown requested..");
             // Channel might already be disconnected if shutdown

radicle-node/src/runtime/handle.rs

@@ -8,6 +8,7 @@ use crossbeam_channel as chan;
 use radicle::node::{ConnectOptions, ConnectResult, Link, Seeds};
 use radicle::storage::refs::RefsAt;
 use reactor::poller::popol::PopolWaker;
+use serde_json::json;
 use thiserror::Error;
 
 use crate::identity::RepoId;
@@ -319,4 +320,45 @@ impl radicle::node::Handle for Handle {
             .shutdown()
             .map_err(|_| Error::ChannelDisconnected)
     }
+
+    fn debug(&self) -> Result<serde_json::Value, Self::Error> {
+        let (sender, receiver) = chan::bounded(1);
+        let query: Arc<QueryState> = Arc::new(move |state| {
+            let debug = serde_json::json!({
+                "outboxSize": state.outbox().len(),
+                "fetching": state.fetching().iter().map(|(rid, state)| {
+                    json!({
+                        "rid": rid,
+                        "from": state.from,
+                        "refsAt": state.refs_at,
+                        "subscribers": state.subscribers.len(),
+                    })
+                }).collect::<Vec<_>>(),
+                "queue": state.queue().iter().map(|fetch| {
+                    json!({
+                        "rid": fetch.rid,
+                        "from": fetch.from,
+                        "refsAt": fetch.refs_at,
+                    })
+                }).collect::<Vec<_>>(),
+                "rateLimiter": state.limiter().buckets.iter().map(|(host, bucket)| {
+                    json!({
+                        "host": host.to_string(),
+                        "bucket": bucket
+                    })
+                }).collect::<Vec<_>>(),
+                "metrics": state.metrics(),
+            });
+            sender.send(debug).ok();
+
+            Ok(())
+        });
+        let (err_sender, err_receiver) = chan::bounded(1);
+        self.command(service::Command::QueryState(query, err_sender))?;
+        err_receiver.recv()??;
+
+        let debug = receiver.recv()?;
+
+        Ok(debug)
+    }
 }

radicle-node/src/service.rs

@@ -108,6 +108,38 @@ pub use message::INVENTORY_LIMIT;
 /// Maximum number of project git references imposed by message size limits.
 pub use message::REF_REMOTE_LIMIT;
 
+/// Metrics we track.
+#[derive(Clone, Debug, Default, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Metrics {
+    peers: HashMap<NodeId, PeerMetrics>,
+}
+
+impl Metrics {
+    /// Get metrics for the given peer.
+    pub fn peer(&mut self, nid: NodeId) -> &mut PeerMetrics {
+        self.peers.entry(nid).or_default()
+    }
+}
+
+/// Per-peer metrics we track.
+#[derive(Clone, Debug, Default, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PeerMetrics {
+    pub received_git_bytes: usize,
+    pub received_fetch_requests: usize,
+    pub received_bytes: usize,
+    pub received_gossip_messages: usize,
+    pub sent_bytes: usize,
+    pub sent_fetch_requests: usize,
+    pub sent_git_bytes: usize,
+    pub sent_gossip_messages: usize,
+    pub streams_opened: usize,
+    pub inbound_connection_attempts: usize,
+    pub outbound_connection_attempts: usize,
+    pub disconnects: usize,
+}
+
 /// Result of syncing our routing table with a node's inventory.
 #[derive(Default)]
 struct SyncedRouting {
@@ -252,13 +284,13 @@ enum TryFetchError<'a> {
 
 /// Fetch state for an ongoing fetch.
 #[derive(Debug)]
-struct FetchState {
+pub struct FetchState {
     /// Node we're fetching from.
-    from: NodeId,
+    pub from: NodeId,
     /// What refs we're fetching.
-    refs_at: Vec<RefsAt>,
+    pub refs_at: Vec<RefsAt>,
     /// Channels waiting for fetch results.
-    subscribers: Vec<chan::Sender<FetchResult>>,
+    pub subscribers: Vec<chan::Sender<FetchResult>>,
 }
 
 impl FetchState {
@@ -272,13 +304,13 @@ impl FetchState {
 
 /// Fetch waiting to be processed, in the fetch queue.
 #[derive(Debug)]
-struct QueuedFetch {
+pub struct QueuedFetch {
     /// Repo being fetched.
-    rid: RepoId,
+    pub rid: RepoId,
     /// Peer being fetched from.
-    from: NodeId,
+    pub from: NodeId,
     /// Refs being fetched.
-    refs_at: Vec<RefsAt>,
+    pub refs_at: Vec<RefsAt>,
     /// The timeout given for the fetch request.
     timeout: time::Duration,
     /// Result channel.
@@ -409,6 +441,8 @@ pub struct Service<D, S, G> {
     emitter: Emitter<Event>,
     /// Local listening addresses.
     listening: Vec<net::SocketAddr>,
+    /// Latest metrics for all nodes connected to since the last start.
+    metrics: Metrics,
 }
 
 impl<D, S, G> Service<D, S, G>
@@ -476,6 +510,7 @@ where
             started_at: None, // Updated on initialize.
             emitter,
             listening: vec![],
+            metrics: Metrics::default(),
         }
     }
 
@@ -695,7 +730,7 @@ where
         Ok(())
     }
 
-    pub fn tick(&mut self, now: LocalTime) {
+    pub fn tick(&mut self, now: LocalTime, metrics: &Metrics) {
         trace!(
             target: "service",
             "Tick +{}",
@@ -712,6 +747,7 @@ where
                 "System clock is not monotonic: {now} is not greater or equal to {}", self.clock
             );
         }
+        self.metrics = metrics.clone();
     }
 
     pub fn wake(&mut self) {
@@ -2391,6 +2427,14 @@ pub trait ServiceState {
     fn nid(&self) -> &NodeId;
     /// Get the existing sessions.
     fn sessions(&self) -> &Sessions;
+    /// Get fetch state.
+    fn fetching(&self) -> &HashMap<RepoId, FetchState>;
+    /// Get fetch queue.
+    fn queue(&self) -> &VecDeque<QueuedFetch>;
+    /// Get outbox.
+    fn outbox(&self) -> &Outbox;
+    /// Get rate limitter.
+    fn limiter(&self) -> &RateLimiter;
     /// Get a repository from storage.
     fn get(&self, rid: RepoId) -> Result<Option<Doc<Verified>>, RepositoryError>;
     /// Get the clock.
@@ -2399,6 +2443,8 @@ pub trait ServiceState {
     fn clock_mut(&mut self) -> &mut LocalTime;
     /// Get service configuration.
     fn config(&self) -> &Config;
+    /// Get service metrics.
+    fn metrics(&self) -> &Metrics;
 }
 
 impl<D, S, G> ServiceState for Service<D, S, G>
@@ -2415,6 +2461,22 @@ where
         &self.sessions
     }
 
+    fn fetching(&self) -> &HashMap<RepoId, FetchState> {
+        &self.fetching
+    }
+
+    fn queue(&self) -> &VecDeque<QueuedFetch> {
+        &self.queue
+    }
+
+    fn outbox(&self) -> &Outbox {
+        &self.outbox
+    }
+
+    fn limiter(&self) -> &RateLimiter {
+        &self.limiter
+    }
+
     fn get(&self, rid: RepoId) -> Result<Option<Doc<Verified>>, RepositoryError> {
         self.storage.get(rid)
     }
@@ -2430,6 +2492,10 @@ where
     fn config(&self) -> &Config {
         &self.config
     }
+
+    fn metrics(&self) -> &Metrics {
+        &self.metrics
+    }
 }
 
 /// Disconnect reason.

radicle-node/src/service/io.rs

@@ -177,6 +177,12 @@ impl Outbox {
         }
     }
 
+    /// Number of items in outbox.
+    #[allow(clippy::len_without_is_empty)]
+    pub fn len(&self) -> usize {
+        self.io.len()
+    }
+
     #[cfg(any(test, feature = "test"))]
     pub(crate) fn queue(&mut self) -> &mut VecDeque<Io> {
         &mut self.io

radicle-node/src/service/limiter.rs

@@ -11,8 +11,8 @@ use radicle::node::{address, config, HostName, NodeId};
 /// bucket's capacity.
 #[derive(Debug, Default)]
 pub struct RateLimiter {
-    buckets: HashMap<HostName, TokenBucket>,
+    pub buckets: HashMap<HostName, TokenBucket>,
-    bypass: HashSet<NodeId>,
+    pub bypass: HashSet<NodeId>,
 }
 
 impl RateLimiter {
@@ -74,7 +74,8 @@ impl AsTokens for config::RateLimit {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
 pub struct TokenBucket {
     /// Token refill rate per second.
     rate: f64,

radicle-node/src/test/handle.rs

@@ -115,4 +115,8 @@ impl radicle::node::Handle for Handle {
     fn shutdown(self) -> Result<(), Self::Error> {
         Ok(())
     }
+
+    fn debug(&self) -> Result<serde_json::Value, Self::Error> {
+        Ok(serde_json::Value::Null)
+    }
 }

radicle-node/src/test/peer.rs

@@ -406,13 +406,15 @@ where
     pub fn messages(&mut self, remote: NodeId) -> impl Iterator<Item = Message> {
         let mut msgs = Vec::new();
 
-        self.service.outbox().queue().retain(|o| match o {
+        Service::outbox(&mut self.service)
-            Io::Write(a, messages) if *a == remote => {
+            .queue()
-                msgs.extend(messages.clone());
+            .retain(|o| match o {
-                false
+                Io::Write(a, messages) if *a == remote => {
-            }
+                    msgs.extend(messages.clone());
-            _ => true,
+                    false
-        });
+                }
+                _ => true,
+            });
 
         msgs.into_iter()
     }
@@ -423,7 +425,7 @@ where
         let mut filtered: Vec<Message> = Vec::new();
         let nid = *self.nid();
 
-        for o in self.service.outbox().queue() {
+        for o in Service::outbox(&mut self.service).queue() {
             match o {
                 Io::Write(a, messages) if *a == remote => {
                     let (relayed, other): (Vec<Message>, _) =
@@ -448,7 +450,7 @@ where
     pub fn inventory_announcements(&mut self, remote: NodeId) -> impl Iterator<Item = Message> {
         let mut invs: Vec<Message> = Vec::new();
 
-        for o in self.service.outbox().queue() {
+        for o in Service::outbox(&mut self.service).queue() {
             match o {
                 Io::Write(a, messages) if *a == remote => {
                     let (inventories, other): (Vec<Message>, _) =
@@ -473,12 +475,12 @@ where
 
     /// Get a draining iterator over the peer's I/O outbox.
     pub fn outbox(&mut self) -> impl Iterator<Item = Io> + '_ {
-        iter::from_fn(|| self.service.outbox().next())
+        iter::from_fn(|| Service::outbox(&mut self.service).next())
     }
 
     /// Get a draining iterator over the peer's I/O outbox, which only returns fetches.
     pub fn fetches(&mut self) -> impl Iterator<Item = (RepoId, NodeId)> + '_ {
-        iter::from_fn(|| self.service.outbox().next()).filter_map(|io| {
+        iter::from_fn(|| Service::outbox(&mut self.service).next()).filter_map(|io| {
             if let Io::Fetch { rid, remote, .. } = io {
                 Some((rid, remote))
             } else {

radicle-node/src/test/simulator.rs

@@ -17,7 +17,7 @@ use log::*;
 use crate::crypto::Signer;
 use crate::prelude::{Address, RepoId};
 use crate::service::io::Io;
-use crate::service::{DisconnectReason, Event, Message, NodeId};
+use crate::service::{DisconnectReason, Event, Message, Metrics, NodeId};
 use crate::storage::Namespaces;
 use crate::storage::{ReadRepository, WriteStorage};
 use crate::test::arbitrary;
@@ -368,7 +368,7 @@ impl<S: WriteStorage + 'static, G: Signer> Simulation<S, G> {
             let Scheduled { input, node, .. } = next;
 
             if let Some(ref mut p) = nodes.get_mut(&node) {
-                p.tick(time);
+                p.tick(time, &Metrics::default());
 
                 match input {
                     Input::Connecting { id, addr } => {

radicle-node/src/wire/protocol.rs

@@ -30,7 +30,7 @@ use crate::prelude::Deserializer;
 use crate::service;
 use crate::service::io::Io;
 use crate::service::FETCH_TIMEOUT;
-use crate::service::{session, DisconnectReason, Service};
+use crate::service::{session, DisconnectReason, Metrics, Service};
 use crate::wire::frame;
 use crate::wire::frame::{Frame, FrameData, StreamId};
 use crate::wire::Encode;
@@ -309,6 +309,8 @@ pub struct Wire<D, S, G: Signer + Ecdh> {
     worker: chan::Sender<Task>,
     /// Used for authentication.
     signer: G,
+    /// Node metrics.
+    metrics: service::Metrics,
     /// Internal queue of actions to send to the reactor.
     actions: VecDeque<Action<G>>,
     /// Outbound attempted peers without a session.
@@ -334,6 +336,7 @@ where
             service,
             worker,
             signer,
+            metrics: Metrics::default(),
             actions: VecDeque::new(),
             inbound: RandomMap::default(),
             outbound: RandomMap::default(),
@@ -456,11 +459,13 @@ where
             log::debug!(target: "wire", "Stream {stream} cannot be found; ignoring flush");
             return;
         };
+        let metrics = self.metrics.peer(remote);
 
         for data in s.channels.try_iter() {
             let frame = match data {
                 ChannelEvent::Data(data) => {
-                    s.sent_bytes += data.len();
+                    metrics.sent_git_bytes += data.len();
+                    metrics.sent_bytes += data.len();
                     Frame::git(stream, data)
                 }
                 ChannelEvent::Close => Frame::control(*link, frame::Control::Close { stream }),
@@ -498,8 +503,10 @@ where
     type Command = Control;
 
     fn tick(&mut self, time: Timestamp) {
-        self.service
+        self.service.tick(
-            .tick(LocalTime::from_millis(time.as_millis() as u128));
+            LocalTime::from_millis(time.as_millis() as u128),
+            &self.metrics,
+        );
     }
 
     fn handle_timer(&mut self) {
@@ -603,6 +610,7 @@ where
                     return;
                 }
                 let (addr, link) = if let Some(peer) = self.inbound.remove(&fd) {
+                    self.metrics.peer(nid).inbound_connection_attempts += 1;
                     (peer.addr, Link::Inbound)
                 } else if let Some(peer) = self.outbound.remove(&fd) {
                     assert_eq!(nid, peer.nid);
@@ -711,6 +719,9 @@ where
                     ..
                 }) = self.peers.get_mut(&id)
                 {
+                    let metrics = self.metrics.peer(*nid);
+                    metrics.received_bytes += data.len();
+
                     if inbox.input(&data).is_err() {
                         log::error!(target: "wire", "Maximum inbox size ({MAX_INBOX_SIZE}) reached for peer {nid}");
                         log::error!(target: "wire", "Unable to process messages fast enough for peer {nid}; disconnecting..");
@@ -726,6 +737,8 @@ where
                                 ..
                             })) => {
                                 log::debug!(target: "wire", "Received `open` command for stream {stream} from {nid}");
+                                metrics.streams_opened += 1;
+                                metrics.received_fetch_requests += 1;
 
                                 let Some(channels) = streams.register(stream, FETCH_TIMEOUT) else {
                                     log::warn!(target: "wire", "Peer attempted to open already-open stream stream {stream}");
@@ -777,6 +790,7 @@ where
                                 data: FrameData::Gossip(msg),
                                 ..
                             })) => {
+                                metrics.received_gossip_messages += 1;
                                 self.service.received_message(*nid, msg);
                             }
                             Ok(Some(Frame {
@@ -785,7 +799,7 @@ where
                                 ..
                             })) => {
                                 if let Some(s) = streams.get_mut(&stream) {
-                                    s.received_bytes += data.len();
+                                    metrics.received_git_bytes += data.len();
 
                                     if s.channels.send(ChannelEvent::Data(data)).is_err() {
                                         log::error!(target: "wire", "Worker is disconnected; cannot send data");
@@ -943,13 +957,17 @@ where
                     log::trace!(
                         target: "wire", "Writing {} message(s) to {}", msgs.len(), node_id
                     );
-
                     let mut data = Vec::new();
+                    let metrics = self.metrics.peer(node_id);
+                    metrics.sent_gossip_messages += msgs.len();
+
                     for msg in msgs {
                         Frame::gossip(link, msg)
                             .encode(&mut data)
                             .expect("in-memory writes never fail");
                     }
+                    metrics.sent_bytes += data.len();
+
                     self.actions.push_back(reactor::Action::Send(fd, data));
                 }
                 Io::Connect(node_id, addr) => {
@@ -963,6 +981,7 @@ where
                         continue;
                     }
                     self.service.attempted(node_id, addr.clone());
+                    self.metrics.peer(node_id).outbound_connection_attempts += 1;
 
                     match dial::<G>(
                         addr.to_inner(),
@@ -1003,7 +1022,9 @@ where
                 }
                 Io::Disconnect(nid, reason) => {
                     if let Some((id, Peer::Connected { .. })) = self.peers.lookup(&nid) {
-                        self.disconnect(id, reason);
+                        if let Some((nid, _)) = self.disconnect(id, reason) {
+                            self.metrics.peer(nid).disconnects += 1;
+                        }
                     } else {
                         log::warn!(target: "wire", "Peer {nid} is not connected: ignoring disconnect");
                     }
@@ -1053,6 +1074,10 @@ where
                     if self.worker.send(task).is_err() {
                         log::error!(target: "wire", "Worker pool is disconnected; cannot send fetch request");
                     }
+                    let metrics = self.metrics.peer(remote);
+                    metrics.streams_opened += 1;
+                    metrics.sent_fetch_requests += 1;
+
                     self.actions.push_back(Action::Send(
                         fd,
                         Frame::control(link, frame::Control::Open { stream }).to_bytes(),

radicle/src/node.rs

@@ -516,6 +516,9 @@ pub enum Command {
     /// Get the node's status.
     Status,
 
+    /// Get node debug information.
+    Debug,
+
     /// Get the node's NID.
     NodeId,
 
@@ -928,6 +931,8 @@ pub trait Handle: Clone + Sync + Send {
     fn sessions(&self) -> Result<Self::Sessions, Self::Error>;
     /// Subscribe to node events.
     fn subscribe(&self, timeout: time::Duration) -> Result<Self::Events, Self::Error>;
+    /// Return debug information as a JSON value.
+    fn debug(&self) -> Result<json::Value, Self::Error>;
 }
 
 /// Iterator of results `T` when passing a [`Command`] to [`Node::call`].
@@ -1239,6 +1244,15 @@ impl Handle for Node {
         Ok(sessions)
     }
 
+    fn debug(&self) -> Result<json::Value, Self::Error> {
+        let debug = self
+            .call::<json::Value>(Command::Debug, DEFAULT_TIMEOUT)?
+            .next()
+            .ok_or(Error::EmptyResponse {})??;
+
+        Ok(debug)
+    }
+
     fn shutdown(self) -> Result<(), Error> {
         for line in self.call::<Success>(Command::Shutdown, DEFAULT_TIMEOUT)? {
             line?;