add traffic counter

This introduces a new wrapper that counts messages and payload bytes per
peer, split by kind (hello, helloReply, sync, ping, pong), each sent,
received and refused.

The traffic counter goes around the firewall, which means messages
dropped outbound are still counted as sent, and messages dropped inbound
don't count as received.

2 files changed, 9 insertions, 0 deletions

diff --git a/src/api/mod.rs b/src/api/mod.rs
index a2bd7e0..bf74de5 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -153,6 +153,9 @@ fn response_for_value(
             peers::PeerListResponse,
             peers::PeerLivenessStatus,
             peers::EpochInfo,
+            crate::transport::traffic::PeerTraffic,
+            crate::transport::traffic::ByKind,
+            crate::transport::traffic::Counter,
         )
     ),
     tags(
diff --git a/src/api/peers.rs b/src/api/peers.rs
index bafb5fa..9ed11c5 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -18,6 +18,7 @@ use crate::document::replica::Epoch;
 use crate::document::store;
 use crate::state::{AppState, HelloState, Liveness, PeerState, unix_millis};
 use crate::swarm::peer::forgotten_path;
+use crate::transport::traffic::PeerTraffic;
 
 #[derive(Debug, Serialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
@@ -65,6 +66,9 @@ pub(crate) struct PeerStatus {
     /// Effective round-trip time in milliseconds (smoothed, raised by requests still
     /// unanswered); absent without a recent sample.
     rtt_ms: Option<f64>,
+    /// Messages and payload bytes exchanged with it, by kind, since we started (or
+    /// since it was last listed).
+    traffic: PeerTraffic,
 }
 
 #[derive(Debug, Serialize, ToSchema)]
@@ -116,6 +120,7 @@ pub(crate) async fn list_peers(
             .lock_rtt()
             .effective(&public_key, now, state.timing.rtt_fresh_for())
             .map(|rtt| rtt.as_secs_f64() * 1000.0);
+        let traffic = state.traffic.peer(&public_key).unwrap_or_default();
         let ps = peer_state_mutex.lock().await;
 
         let (status, dormant_for_secs) = match ps.liveness {
@@ -141,6 +146,7 @@ pub(crate) async fn list_peers(
             heads: ps.reported_heads,
             neighbour,
             rtt_ms,
+            traffic,
         });
     }