From 5854fe5a3a7c7e8f681264d3a4b27597a8c61de0 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 20:48:45 +0100 Subject: add sync and relay stats --- src/api/stats.rs | 49 ++++++++++++++++++++++++++++++-- src/api/stats/prometheus.rs | 68 +++++++++++++++++++++++++++++++++++++++++++++ src/swarm/blob.rs | 1 + src/swarm/relay.rs | 10 ++++++- src/swarm/stats.rs | 34 +++++++++++++++++++++++ 5 files changed, 158 insertions(+), 4 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, } +/// 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, + /// 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 = 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, + last_sync_secs: Option, + diverged_secs: Option, traffic: PeerTraffic, incidents: Incidents, rtt_secs: Option, @@ -142,6 +144,12 @@ async fn peers(state: &AppState) -> Vec { 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", @@ -489,6 +537,26 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { out.sample("peer_last_seen_seconds", &[("peer", &peer.key)], secs); } } + 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", diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index a2feb98..d68a78c 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -124,6 +124,7 @@ pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) { match message.body { BlobBody::Request { chunks } => { let Ok(permit) = Arc::clone(&state.blobs.answering).try_acquire_owned() else { + state.stats.blob_request_dropped(); debug!(peer = %peer, blob = %message.id, "Answering too many blob requests; dropping one"); return; }; diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 087e028..2fa5f5f 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -81,6 +81,7 @@ pub(crate) async fn announce(state: &AppState, from: &str) { /// round with a single sender skips it; with several, each lacks the others' changes, /// so every neighbour is told. async fn announce_now(state: &AppState, senders: &HashSet) { + state.stats.announce_round(); let skip = match senders.len() { 1 => senders.iter().next().cloned(), _ => None, @@ -165,7 +166,13 @@ pub(crate) async fn pull_if_behind( Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT ); if busy { - pulling.waiting.insert(peer.to_owned(), remote.clone()); + if pulling + .waiting + .insert(peer.to_owned(), remote.clone()) + .is_none() + { + state.stats.pull_queued(); + } return false; } pulling.current = Some((peer.to_owned(), now)); @@ -174,6 +181,7 @@ pub(crate) async fn pull_if_behind( // Automerge asks for whatever these heads need that we don't have. peer_state.lock().await.sync.pull_from(theirs); + state.stats.pull_started(); debug!(peer = %peer, "Neighbour has changes we lack; pulling"); let _ = state .sync_events diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index f9a5b7b..8172233 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -29,6 +29,8 @@ struct Inner { catch_up_sources: u32, max_heads: usize, stale_hello_replies: u64, + relay: Relay, + blob_requests_dropped: u64, saved_size: Option, measuring: bool, /// The document version (`AppState::change_tx`) and epoch the saved size is of. @@ -90,6 +92,17 @@ pub(crate) struct SavedSize { pub(crate) took: Duration, } +/// Passing changes on (`swarm::relay`), since start. +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct Relay { + /// Announcement rounds (each a PING to the neighbours). + pub(crate) announce_rounds: u64, + /// Pulls started. + pub(crate) pulls: u64, + /// Announcements of heads we lacked that had to wait for the pull in progress. + pub(crate) pulls_queued: u64, +} + /// When the initial sync finished, and what it took. #[derive(Debug, Clone, Copy)] pub(crate) struct CaughtUp { @@ -107,6 +120,8 @@ pub(crate) struct Snapshot { pub(crate) catch_up_sources: u32, pub(crate) max_heads: usize, pub(crate) stale_hello_replies: u64, + pub(crate) relay: Relay, + pub(crate) blob_requests_dropped: u64, pub(crate) propagation: Propagation, } @@ -154,6 +169,23 @@ impl NodeStats { self.lock().stale_hello_replies += 1; } + pub(crate) fn announce_round(&self) { + self.lock().relay.announce_rounds += 1; + } + + pub(crate) fn pull_started(&self) { + self.lock().relay.pulls += 1; + } + + pub(crate) fn pull_queued(&self) { + self.lock().relay.pulls_queued += 1; + } + + /// A blob request went unanswered: too many were being answered. + pub(crate) fn blob_request_dropped(&self) { + self.lock().blob_requests_dropped += 1; + } + /// The last measurement of the saved size, if any. pub(crate) fn saved_size(&self) -> Option { self.lock().saved_size @@ -203,6 +235,8 @@ impl NodeStats { catch_up_sources: inner.catch_up_sources, max_heads: inner.max_heads, stale_hello_replies: inner.stale_hello_replies, + relay: inner.relay, + blob_requests_dropped: inner.blob_requests_dropped, propagation: inner.propagation, } } -- cgit v1.3.1