1 files changed, 67 insertions, 8 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
index d60e7b3..7caaaf3 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -3,9 +3,9 @@
 use std::sync::Arc;
 
 use axum::{
+    Json,
     extract::{Path, Query, State},
     http::StatusCode,
-    Json,
 };
 use serde::{Deserialize, Serialize};
 use tokio::sync::Mutex;
@@ -16,8 +16,9 @@ use utoipa::ToSchema;
 use super::store_status;
 use crate::document::replica::Epoch;
 use crate::document::store;
-use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState};
+use crate::state::{AppState, HelloState, Incidents, Liveness, PeerState, unix_millis};
 use crate::swarm::peer::forgotten_path;
+use crate::transport::traffic::PeerTraffic;
 
 #[derive(Debug, Serialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
@@ -47,6 +48,8 @@ impl From<Epoch> for EpochInfo {
 #[serde(rename_all = "camelCase")]
 pub(crate) struct PeerStatus {
     public_key: String,
+    /// The name the peer gave itself in the member list, if any (not unique).
+    alias: Option<String>,
     validated_swarm: Option<String>,
     status: PeerLivenessStatus,
     handshake_complete: bool,
@@ -57,12 +60,41 @@ pub(crate) struct PeerStatus {
     epoch: Option<EpochInfo>,
     /// Whether that epoch equals ours, i.e. whether we sync with this peer.
     same_epoch: Option<bool>,
+    /// Number of heads in its last keepalive; absent before one, or when it had more
+    /// than 32 (then only their digest is sent).
+    heads: Option<usize>,
+    /// We keep it alive and sync with it on our own; other peers we only answer.
+    neighbour: bool,
+    /// 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,
+    incidents: Incidents,
+    /// Smoothed sync throughput (`swarm::throughput`); absent before the first large
+    /// exchange.
+    sync_throughput: Option<ThroughputInfo>,
+}
+
+#[derive(Debug, Serialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct ThroughputInfo {
+    bytes_per_sec: f64,
+    samples: u32,
+    /// Age of the last sample.
+    secs_ago: u64,
 }
 
 #[derive(Debug, Serialize, ToSchema)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct PeerListResponse {
+    /// Our own alias (`--alias`).
+    local_alias: Option<String>,
     local_epoch: EpochInfo,
+    /// Number of heads of our document. Several heads are concurrent changes no later
+    /// change has merged yet.
+    local_heads: usize,
     peers: Vec<PeerStatus>,
 }
 
@@ -97,9 +129,26 @@ pub(crate) async fn list_peers(
 
     let now = Instant::now();
     let local_epoch = state.local_epoch();
+    let mut aliases = crate::swarm::members::aliases(&state.doc.read().await.doc);
     let mut peers = Vec::with_capacity(snapshot.len());
 
     for (public_key, peer_state_mutex) in snapshot {
+        let neighbour = crate::swarm::topology::is_neighbour(&state, &public_key);
+        let rtt_ms = state
+            .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 sync_throughput = state
+            .throughput
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .get(&public_key)
+            .map(|estimate| ThroughputInfo {
+                bytes_per_sec: estimate.bytes_per_sec.round(),
+                samples: estimate.samples,
+                secs_ago: now.saturating_duration_since(estimate.at).as_secs(),
+            });
         let ps = peer_state_mutex.lock().await;
 
         let (status, dormant_for_secs) = match ps.liveness {
@@ -111,26 +160,33 @@ pub(crate) async fn list_peers(
         };
 
         peers.push(PeerStatus {
+            alias: aliases.remove(&public_key),
             public_key,
-            validated_swarm: match &ps.membership {
-                PeerMembership::Unknown => None,
-                PeerMembership::Member { swarm_id } => Some(swarm_id.clone()),
-            },
+            validated_swarm: ps.standing.swarm_id().map(str::to_owned),
             status,
             handshake_complete: ps.hello_state == HelloState::Established,
             last_seen_secs_ago: ps
                 .last_seen
                 .map(|t| now.saturating_duration_since(t).as_secs()),
             dormant_for_secs,
-            has_pending_delivery: ps.pending_outbound_sync.is_some(),
+            has_pending_delivery: ps.sync.sending.is_some() || ps.sync.retry_at.is_some(),
             epoch: ps.epoch.map(EpochInfo::from),
             same_epoch: ps.epoch.map(|epoch| epoch == local_epoch),
+            heads: ps.reported_heads,
+            neighbour,
+            rtt_ms,
+            traffic,
+            incidents: ps.incidents,
+            sync_throughput,
         });
     }
 
     peers.sort_by(|a, b| a.public_key.cmp(&b.public_key));
+    let local_heads = state.doc.write().await.doc.get_heads().len();
     Ok(Json(PeerListResponse {
+        local_alias: state.alias.clone(),
         local_epoch: local_epoch.into(),
+        local_heads,
         peers,
     }))
 }
@@ -178,7 +234,10 @@ pub(crate) async fn delete_peer(
 
     let peer_clone = peer.clone();
     state
-        .mutate_doc(move |doc| store::write_path(doc, &forgotten_path(&peer_clone), &record))
+        .mutate_doc(move |doc| {
+            store::write_path(doc, &forgotten_path(&peer_clone), &record)?;
+            crate::swarm::members::unregister(doc, &peer_clone)
+        })
         .await
         .map_err(|e| store_status(e, "Failed to write forget record"))?;