add per-node traffic counts

4 files changed, 67 insertions, 5 deletions

diff --git a/docs/measurements.md b/docs/measurements.md
index 2adc0f5..fb01de0 100644
--- a/docs/measurements.md
+++ b/docs/measurements.md
@@ -168,9 +168,9 @@ clocks"); setting them is the operating system's job (NTP).
 
 Messages and payload bytes by direction (`sent`, `received`, `refused`) and kind
 (`hello`, `helloReply`, `sync`, `status`, `statusReply`, `blob`, `identity` for WHO and
-ME, counted under the address they went to or came from, so only in the totals, and
-`paths` for PING and PONG: per peer, but the paths layer's own PINGs under the address,
-so only in the totals). Payload bytes are what a transport's message carries (the wire
+ME, and `paths` for PING and PONG). The layers below the node count WHO and ME, and the
+paths layer's own PINGs and PONGs, under the address they went to or came from; a peer
+(`/peers`) has those of its addresses too, where it is known to be at them. Payload bytes are what a transport's message carries (the wire
 encoding): not the transport's framing (for Mycelium, its topic and the base64 of its
 HTTP API). `refused` is what the transports wouldn't accept to send (no path to the
 peer), not counted as sent.
diff --git a/src/api/peers.rs b/src/api/peers.rs
index 3fe4c39..537ad98 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -121,7 +121,9 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse {
 
     for (peer, peer_state_mutex) in snapshot {
         let neighbour = crate::swarm::topology::is_neighbour(&state, &peer);
-        let traffic = state.traffic.peer(&peer).unwrap_or_default();
+        let traffic = state
+            .traffic
+            .node(&peer, &state.transport.identities_of(&peer));
         let sync_throughput = state
             .throughput
             .lock_or_recover()
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index e8e6f8b..6196571 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -170,7 +170,15 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             crate::swarm::state_file::update(&state, false).await;
             {
                 let known = state.peer_states.read().await;
-                state.traffic.retain(|peer| known.contains_key(peer));
+                // And their addresses, where the layers below the node count
+                // (`Traffic::node`).
+                let addresses: std::collections::HashSet<String> = known
+                    .keys()
+                    .flat_map(|peer| state.transport.identities_of(peer))
+                    .collect();
+                state
+                    .traffic
+                    .retain(|peer| known.contains_key(peer) || addresses.contains(peer));
                 state
                     .throughput
                     .lock_or_recover()
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index bf9438f..b7907fb 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -30,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)]
@@ -51,6 +56,37 @@ pub(crate) struct ByKind {
 }
 
 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,
@@ -282,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
     }