path: root/src/api/peers.rs

//! /peers handlers.

use std::sync::Arc;

use axum::{
    Json,
    extract::{Path, Query, State},
    http::StatusCode,
};
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
use tokio::time::Instant;
use tracing::warn;
use utoipa::ToSchema;

use super::store_status;
use crate::document::replica::Epoch;
use crate::document::store;
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")]
pub(crate) enum PeerLivenessStatus {
    Active,
    Dormant,
}

#[derive(Debug, Serialize, ToSchema)]
#[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, ToSchema)]
#[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,
    last_seen_secs_ago: Option<u64>,
    dormant_for_secs: Option<u64>,
    has_pending_delivery: bool,
    /// Epoch the peer last advertised; absent until it has sent a message.
    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>,
}

#[derive(Debug, Deserialize)]
pub(crate) struct DeletePeerQuery {
    #[serde(default)]
    force: bool,
}

#[utoipa::path(
    get,
    path = "/peers",
    tag = "Peers",
    responses(
        (
            status = 200,
            description = "Lists known peers and their local replication status",
            body = PeerListResponse
        )
    )
)]
pub(crate) async fn list_peers(
    State(state): State<AppState>,
) -> Result<Json<PeerListResponse>, StatusCode> {
    let snapshot: Vec<(String, Arc<Mutex<PeerState>>)> = state
        .peer_states
        .read()
        .await
        .iter()
        .map(|(pk, m)| (pk.clone(), Arc::clone(m)))
        .collect();

    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 {
            Liveness::Active => (PeerLivenessStatus::Active, None),
            Liveness::Dormant { since } => (
                PeerLivenessStatus::Dormant,
                Some(now.saturating_duration_since(since).as_secs()),
            ),
        };

        peers.push(PeerStatus {
            alias: aliases.remove(&public_key),
            public_key,
            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.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.write().await.doc.get_heads().len();
    Ok(Json(PeerListResponse {
        local_alias: state.alias.clone(),
        local_epoch: local_epoch.into(),
        local_heads,
        peers,
    }))
}

#[utoipa::path(
    delete,
    path = "/peers/{peer}",
    tag = "Peers",
    params(
        ("peer" = String, Path, description = "Peer public key"),
        ("force" = Option<bool>, Query, description = "Also forget a peer that is currently alive; it returns on its next message")
    ),
    responses(
        (status = 204, description = "Swarm-wide forget recorded"),
        (status = 400, description = "Not a public key, or the local node's own key"),
        (status = 409, description = "Peer is alive; pass force=true"),
        (status = 503, description = "Node is not ready")
    )
)]
pub(crate) async fn delete_peer(
    State(state): State<AppState>,
    Path(peer): Path<String>,
    Query(query): Query<DeletePeerQuery>,
) -> Result<StatusCode, StatusCode> {
    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE);
    }
    if peer.is_empty() || peer == state.local_id() {
        return Err(StatusCode::BAD_REQUEST);
    }

    // Unknown locally is fine: other nodes may still know the peer.
    if let Some(peer_state_mutex) = state.get_peer_state(&peer).await {
        if peer_state_mutex
            .lock()
            .await
            .is_alive(&state.timing, Instant::now())
            && !query.force
        {
            return Err(StatusCode::CONFLICT);
        }
    }

    let record = serde_json::json!({ "at": unix_millis(), "by": state.local_id() });

    let peer_clone = peer.clone();
    state
        .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"))?;

    if state.peer_states.write().await.remove(&peer).is_some() {
        warn!(peer = %peer, "Peer forgotten swarm-wide by operator");
    }

    Ok(StatusCode::NO_CONTENT)
}