add sync and relay stats

2 files changed, 114 insertions, 3 deletions

diff --git a/src/api/stats.rs b/src/api/stats.rs
index 5d32d82..a34d4ed 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -29,6 +29,8 @@ pub(crate) struct StatsResponse {
     document: DocumentStats,
     catch_up: CatchUpStats,
     remakes: RemakeStats,
+    sync: SyncStats,
+    relay: RelayStats,
     topology: TopologyStats,
     stability: StabilityStats,
     /// HELLO_REPLYs to an earlier run of ours, ignored.
@@ -103,6 +105,29 @@ struct CatchUpStats {
     sync_bytes: Option<u64>,
 }
 
+/// How sync with the neighbours goes.
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct SyncStats {
+    /// Since a sync message from any neighbour was applied; absent if none ever was. A
+    /// node whose writes go nowhere shows it growing.
+    last_from_neighbour_secs: Option<u64>,
+    /// Peers whose heads differ from ours while sync is idle (`peers.md`, healing).
+    diverged_peers: usize,
+}
+
+/// Passing changes on (`swarm::relay`) and serving blobs, since start.
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct RelayStats {
+    announce_rounds: u64,
+    pulls: u64,
+    /// Announcements of heads we lacked that waited for the pull in progress.
+    pulls_queued: u64,
+    /// Blob requests not answered: too many were being answered.
+    blob_requests_dropped: u64,
+}
+
 /// Remakes and their aftermath here (`remake::Summary`).
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -186,9 +211,17 @@ async fn collect(state: &AppState) -> StatsResponse {
     let (connected_fraction, observed) = stability::measured(state);
 
     let mut incidents = Incidents::default();
-    let peers: Vec<_> = state.peer_states.read().await.values().cloned().collect();
-    for peer_state in peers {
-        incidents.add(&peer_state.lock().await.incidents);
+    let mut last_from_neighbour: Option<tokio::time::Instant> = 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() {
+            diverged_peers += 1;
+        }
+        if topology::is_neighbour(state, &peer) {
+            last_from_neighbour = last_from_neighbour.max(peer_state.last_sync_rx);
+        }
     }
 
     StatsResponse {
@@ -220,6 +253,16 @@ async fn collect(state: &AppState) -> StatsResponse {
                 .map(|created| now_ms.ms_since(created) / 1000),
             archived_epochs: remakes.archived_epochs,
         },
+        sync: SyncStats {
+            last_from_neighbour_secs: last_from_neighbour.map(|at| at.elapsed().as_secs()),
+            diverged_peers,
+        },
+        relay: RelayStats {
+            announce_rounds: node.relay.announce_rounds,
+            pulls: node.relay.pulls,
+            pulls_queued: node.relay.pulls_queued,
+            blob_requests_dropped: node.blob_requests_dropped,
+        },
         topology: TopologyStats {
             k: state.neighbours_k,
             view: topology::summary(state).map(|s| TopologyView {
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 6fd37a0..3baa125 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -112,6 +112,8 @@ struct Peer {
     neighbour: bool,
     dormant: bool,
     last_seen_secs: Option<f64>,
+    last_sync_secs: Option<f64>,
+    diverged_secs: Option<f64>,
     traffic: PeerTraffic,
     incidents: Incidents,
     rtt_secs: Option<f64>,
@@ -142,6 +144,12 @@ async fn peers(state: &AppState) -> Vec<Peer> {
             last_seen_secs: peer_state
                 .last_seen
                 .map(|t| now.saturating_duration_since(t).as_secs_f64()),
+            last_sync_secs: peer_state
+                .last_sync_rx
+                .map(|t| now.saturating_duration_since(t).as_secs_f64()),
+            diverged_secs: peer_state
+                .diverged_since
+                .map(|t| now.saturating_duration_since(t).as_secs_f64()),
             traffic,
             incidents: peer_state.incidents,
             rtt_secs,
@@ -303,6 +311,46 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
         remakes.archived_epochs as f64,
     );
 
+    if let Some(secs) = stats.sync.last_from_neighbour_secs {
+        out.single(
+            "last_sync_from_neighbour_seconds",
+            "gauge",
+            "Seconds since a sync message from any neighbour was applied.",
+            secs as f64,
+        );
+    }
+    out.single(
+        "diverged_peers",
+        "gauge",
+        "Peers whose heads differ from ours while sync is idle.",
+        stats.sync.diverged_peers as f64,
+    );
+    let relay = &stats.relay;
+    out.single(
+        "announce_rounds_total",
+        "counter",
+        "Announcement rounds (our new heads to the neighbours).",
+        relay.announce_rounds as f64,
+    );
+    out.single(
+        "pulls_total",
+        "counter",
+        "Pulls of changes we lacked, started.",
+        relay.pulls as f64,
+    );
+    out.single(
+        "pulls_queued_total",
+        "counter",
+        "Announcements of heads we lacked that waited for the pull in progress.",
+        relay.pulls_queued as f64,
+    );
+    out.single(
+        "blob_requests_dropped_total",
+        "counter",
+        "Blob requests not answered: too many were being answered.",
+        relay.blob_requests_dropped as f64,
+    );
+
     let topology = &stats.topology;
     out.single(
         "neighbours_target",
@@ -490,6 +538,26 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
         }
     }
     out.family(
+        "peer_last_sync_seconds",
+        "gauge",
+        "Seconds since a sync message from the peer was applied (or the epoch changed).",
+    );
+    for peer in &peers {
+        if let Some(secs) = peer.last_sync_secs {
+            out.sample("peer_last_sync_seconds", &[("peer", &peer.key)], secs);
+        }
+    }
+    out.family(
+        "peer_diverged_seconds",
+        "gauge",
+        "Seconds the peer's heads have differed from ours while sync was idle.",
+    );
+    for peer in &peers {
+        if let Some(secs) = peer.diverged_secs {
+            out.sample("peer_diverged_seconds", &[("peer", &peer.key)], secs);
+        }
+    }
+    out.family(
         "peer_rtt_seconds",
         "gauge",
         "Effective round-trip time (smoothed, raised by unanswered requests).",