summaryrefslogtreecommitdiffstats
path: root/src/api/peers.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/api/peers.rs')
-rw-r--r--src/api/peers.rs73
1 files changed, 66 insertions, 7 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs
index 65f8fa1..919f187 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -16,8 +16,9 @@ use utoipa::ToSchema;
use super::store_status;
use crate::document::replica::Epoch;
use crate::document::store;
-use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState, unix_millis};
+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.read().await.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"))?;