1 files changed, 55 insertions, 47 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
index eeb3fb2..537ad98 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -12,12 +12,13 @@ use tokio::time::Instant;
 use tracing::warn;
 
 use super::store_status;
-use crate::document::replica::Epoch;
+use crate::document::replica::EpochRecord;
 use crate::document::store;
 use crate::locks::MutexExt;
 use crate::state::{AppState, HelloState, Incidents, Liveness};
 use crate::swarm::peer::forgotten_path;
 use crate::transport::Destination;
+use crate::transport::paths::ClockOffset;
 use crate::transport::traffic::PeerTraffic;
 
 #[derive(Debug, Serialize)]
@@ -31,77 +32,68 @@ pub(crate) enum PeerLivenessStatus {
 
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
-pub(crate) struct EpochInfo {
-    generation: u64,
-    /// Hex-encoded snapshot hash; all zeros for the genesis epoch.
-    id: String,
-}
-
-impl From<Epoch> for EpochInfo {
-    fn from(epoch: Epoch) -> Self {
-        Self {
-            generation: epoch.generation,
-            id: epoch.id_hex(),
-        }
-    }
-}
-
-#[derive(Debug, Serialize)]
-#[serde(rename_all = "camelCase")]
 pub(crate) struct PeerStatus {
     /// The name the peer gave itself in the member list, if any (not unique).
-    alias: Option<String>,
+    pub(crate) alias: Option<String>,
     /// Where this node reaches it, or would (`Transport::addresses_of`).
-    addresses: Vec<String>,
+    pub(crate) addresses: Vec<String>,
     /// Where messages to it go now: the address, whether the peer proved itself there,
     /// and the connection there (libp2p: direct or relayed). Absent if nowhere.
-    destination: Option<Destination>,
-    validated_swarm: Option<String>,
-    status: PeerLivenessStatus,
-    handshake_complete: bool,
-    last_seen_secs_ago: Option<u64>,
-    dormant_for_secs: Option<u64>,
-    has_pending_delivery: bool,
+    pub(crate) destination: Option<Destination>,
+    pub(crate) validated_swarm: Option<String>,
+    pub(crate) status: PeerLivenessStatus,
+    pub(crate) handshake_complete: bool,
+    pub(crate) last_seen_secs_ago: Option<u64>,
+    pub(crate) dormant_for_secs: Option<u64>,
+    /// Since a sync message from it was applied (or the epoch changed).
+    pub(crate) last_sync_secs_ago: Option<u64>,
+    /// How long its heads have differed from ours with no sync message from it in
+    /// between; absent while they match.
+    pub(crate) diverged_for_secs: Option<u64>,
+    pub(crate) has_pending_delivery: bool,
     /// Epoch the peer last advertised; absent until it has sent a message.
-    epoch: Option<EpochInfo>,
+    pub(crate) epoch: Option<EpochRecord>,
     /// Whether that epoch equals ours, i.e. whether we sync with this peer.
-    same_epoch: Option<bool>,
+    pub(crate) 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 and some of them are sent).
-    heads: Option<usize>,
+    pub(crate) heads: Option<usize>,
     /// We keep it alive and sync with it on our own; other peers we only answer.
-    neighbour: bool,
+    pub(crate) neighbour: bool,
     /// Messages and payload bytes exchanged with it, by kind, since we started (or
     /// since it was last listed).
-    traffic: PeerTraffic,
-    incidents: Incidents,
+    pub(crate) traffic: PeerTraffic,
+    pub(crate) incidents: Incidents,
     /// Smoothed sync throughput (`swarm::throughput`); absent before the first large
     /// exchange.
-    sync_throughput: Option<ThroughputInfo>,
+    pub(crate) sync_throughput: Option<ThroughputInfo>,
+    /// How far its clock is ahead of ours, give or take (`Transport::clock_offset`);
+    /// absent before a PONG from it.
+    pub(crate) clock: Option<ClockOffset>,
 }
 
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct ThroughputInfo {
-    bytes_per_sec: f64,
-    samples: u32,
+    pub(crate) bytes_per_sec: f64,
+    pub(crate) samples: u32,
     /// Age of the last sample.
-    secs_ago: u64,
+    pub(crate) secs_ago: u64,
 }
 
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct PeerListResponse {
     /// Our own alias (`--alias`).
-    local_alias: Option<String>,
-    local_epoch: EpochInfo,
+    pub(crate) local_alias: Option<String>,
+    pub(crate) local_epoch: EpochRecord,
     /// Number of heads of our document. Several heads are concurrent changes no later
     /// change has merged yet.
-    local_heads: usize,
+    pub(crate) local_heads: usize,
     /// Where this node can be reached (`Transport::local_addresses`).
-    local_addresses: Vec<String>,
+    pub(crate) local_addresses: Vec<String>,
     /// By node ID.
-    peers: BTreeMap<String, PeerStatus>,
+    pub(crate) peers: BTreeMap<String, PeerStatus>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -114,6 +106,12 @@ pub(crate) struct DeletePeerQuery {
 pub(crate) async fn list_peers(
     State(state): State<AppState>,
 ) -> Result<Json<PeerListResponse>, StatusCode> {
+    Ok(Json(peer_list(&state).await))
+}
+
+/// What `GET /peers` answers, for others to show too (the dashboard).
+pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse {
+    let state = state.clone();
     let snapshot = state.peers_snapshot().await;
 
     let now = Instant::now();
@@ -123,7 +121,9 @@ pub(crate) async fn list_peers(
 
     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()
@@ -147,6 +147,7 @@ pub(crate) async fn list_peers(
         let alias = aliases.remove(&peer);
         let addresses = state.transport.addresses_of(&peer);
         let destination = state.transport.destination(&peer);
+        let clock = state.transport.clock_offset(&peer);
         peers.insert(
             peer,
             PeerStatus {
@@ -160,26 +161,33 @@ pub(crate) async fn list_peers(
                     .last_seen
                     .map(|t| now.saturating_duration_since(t).as_secs()),
                 dormant_for_secs,
+                last_sync_secs_ago: ps
+                    .last_sync_rx
+                    .map(|t| now.saturating_duration_since(t).as_secs()),
+                diverged_for_secs: ps
+                    .diverged_since
+                    .map(|t| now.saturating_duration_since(t).as_secs()),
                 has_pending_delivery: ps.sync.sending.is_some() || ps.sync.retry_at.is_some(),
-                epoch: ps.epoch.map(EpochInfo::from),
+                epoch: ps.epoch.map(EpochRecord::from),
                 same_epoch: ps.epoch.map(|epoch| epoch == local_epoch),
                 heads: ps.reported_heads,
                 neighbour,
                 traffic,
                 incidents: ps.incidents,
                 sync_throughput,
+                clock,
             },
         );
     }
 
     let local_heads = state.doc.read().await.heads().len();
-    Ok(Json(PeerListResponse {
+    PeerListResponse {
         local_alias: state.alias.clone(),
         local_addresses: state.transport.local_addresses(),
         local_epoch: local_epoch.into(),
         local_heads,
         peers,
-    }))
+    }
 }
 
 /// `DELETE /peers/{pk}`: forgets a peer swarm-wide.
@@ -218,7 +226,7 @@ pub(crate) async fn delete_peer(
             crate::swarm::members::unregister(doc, &peer_clone)
         })
         .await
-        .map_err(|e| store_status(e, "Failed to write forget record"))?;
+        .map_err(|e| store_status(e, "Failed to write forget record").status())?;
 
     // Unless forced, a peer that came alive since the check above stays: it cancels
     // the record.