1 files changed, 35 insertions, 15 deletions

diff --git a/src/api/stats.rs b/src/api/stats.rs
index 889b675..558507a 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -13,8 +13,10 @@ use std::collections::BTreeMap;
 
 use serde::Serialize;
 
-use super::peers::EpochInfo;
+use super::peers::{PeerListResponse, PeerLivenessStatus, peer_list};
+use crate::document::replica::EpochRecord;
 use crate::state::{AppState, Incidents};
+use crate::swarm::skew::{self, Skew};
 use crate::swarm::{members, stability, topology};
 use crate::transport::stats::TransportStats;
 use crate::transport::traffic::PeerTraffic;
@@ -28,7 +30,7 @@ pub(crate) struct StatsResponse {
     alias: Option<String>,
     uptime_secs: u64,
     ready: bool,
-    epoch: EpochInfo,
+    epoch: EpochRecord,
     document: DocumentStats,
     catch_up: CatchUpStats,
     remakes: RemakeStats,
@@ -43,6 +45,8 @@ pub(crate) struct StatsResponse {
     /// The peers listed in /peers together (a forgotten peer's are gone).
     incidents: Incidents,
     propagation: PropagationStats,
+    /// This node's clock against its active peers' (`swarm::skew`).
+    clock: Skew,
     /// This node's own files (`files`), by kind.
     disk: BTreeMap<&'static str, DiskStats>,
     /// How the transports find and reach peers (`Transport::stats`).
@@ -63,13 +67,14 @@ struct DiskStats {
 }
 
 /// How long changes written elsewhere took from their commit there until applied here,
-/// once caught up. Includes the clock skew between the writer and us.
+/// once caught up, by the peers' clocks (`swarm::skew`). Includes the writer's clock
+/// skew.
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct PropagationStats {
     count: u64,
     sum_ms: u64,
-    /// Changes stamped later than our clock on arrival (writer's clock ahead).
+    /// Changes stamped later than their arrival (writer's clock ahead).
     ahead: u64,
     /// Not cumulative; the last bucket has no bound.
     buckets: Vec<PropagationBucket>,
@@ -237,19 +242,22 @@ pub(crate) async fn get_relay_only_stats(
 
 /// `GET /admin/stats`.
 pub(crate) async fn get_stats(State(state): State<AppState>, headers: HeaderMap) -> Response {
-    let stats = collect(&state).await;
+    let list = peer_list(&state).await;
+    let stats = collect(&state, &list).await;
     let accept = headers
         .get(header::ACCEPT)
         .and_then(|value| value.to_str().ok());
     if prometheus::wanted(accept) {
-        let body = prometheus::render(&state, &stats).await;
+        let body = prometheus::render(&stats, &list);
         ([(header::CONTENT_TYPE, prometheus::CONTENT_TYPE)], body).into_response()
     } else {
         Json(stats).into_response()
     }
 }
 
-async fn collect(state: &AppState) -> StatsResponse {
+/// What `GET /admin/stats` answers as JSON, for others to show too (the dashboard); its
+/// totals over the peers from their list (`peer_list`), read once for both.
+pub(crate) async fn collect(state: &AppState, list: &PeerListResponse) -> StatsResponse {
     let node = state.stats.snapshot();
 
     let (heads, tier, remakes) = {
@@ -267,17 +275,28 @@ async fn collect(state: &AppState) -> StatsResponse {
 
     let (connected_fraction, observed) = stability::measured(state);
 
+    let clock = skew::of(
+        &list
+            .peers
+            .values()
+            .filter(|peer| !matches!(peer.status, PeerLivenessStatus::Dormant))
+            .filter_map(|peer| peer.clock)
+            .collect::<Vec<_>>(),
+        state.clock_skew_warn,
+    );
+
     let mut incidents = Incidents::default();
-    let mut last_from_neighbour: Option<tokio::time::Instant> = None;
+    let mut last_from_neighbour_secs: Option<u64> = None;
     let mut diverged_peers = 0;
-    for (peer, peer_state) in state.peers_snapshot().await {
-        let peer_state = peer_state.lock().await;
-        incidents.add(&peer_state.incidents);
-        if peer_state.diverged_since.is_some() {
+    for peer in list.peers.values() {
+        incidents.add(&peer.incidents);
+        if peer.diverged_for_secs.is_some() {
             diverged_peers += 1;
         }
-        if topology::is_neighbour(state, &peer) {
-            last_from_neighbour = last_from_neighbour.max(peer_state.last_sync_rx);
+        if peer.neighbour
+            && let Some(secs) = peer.last_sync_secs_ago
+        {
+            last_from_neighbour_secs = Some(last_from_neighbour_secs.map_or(secs, |s| s.min(secs)));
         }
     }
 
@@ -313,7 +332,7 @@ async fn collect(state: &AppState) -> StatsResponse {
             archived_epochs: remakes.archived_epochs,
         },
         sync: SyncStats {
-            last_from_neighbour_secs: last_from_neighbour.map(|at| at.elapsed().as_secs()),
+            last_from_neighbour_secs,
             diverged_peers,
             largest_received_bytes: node.largest_sync_received,
         },
@@ -342,6 +361,7 @@ async fn collect(state: &AppState) -> StatsResponse {
         stale_hello_replies: node.stale_hello_replies,
         traffic: state.traffic.total(),
         incidents,
+        clock,
         propagation: PropagationStats {
             count: node.propagation.count,
             sum_ms: node.propagation.sum_ms,