ferrix/crates/server/src/session.rs

//! The sessions as the rest of the server sees them: the registry of
//! those that are established, the configured peers and where the dialer
//! is with each, the events a session's coming and going raise, and the
//! word that the daemon is leaving.
//!
//! A session registers itself when it reaches Established and is removed when
//! its [`SessionGuard`] drops, so a session that errors out is always
//! deregistered. Other parts of the server (the advertiser, the HTTP API) use
//! the registry to enumerate live sessions and queue outbound messages.

use std::collections::HashMap;
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use std::time::{Duration, SystemTime};

use bgp_packet::message::Message;
use bgp_primitives::ip_prefix::Nlri;
use bgp_rib::table::PeerIndex;
use bytes::BytesMut;
use tokio::sync::{broadcast, mpsc, watch};

use crate::address::{Endpoint, PeerAddress, socket_text};
use crate::collision::Direction;
use crate::error::SessionError;
use crate::family::FamilySet;
use crate::fsm::Negotiated;
use crate::metrics::{PEER_UP, SESSIONS_ESTABLISHED};
use crate::policy::{ExportLimits, ExportPolicy, PeerRole};
use crate::restart::{Agreed, Leave};
use crate::rpki::Subject;
use crate::sync::IgnorePoisonRw;

const EVENT_CHANNEL_CAPACITY: usize = 256;

/// Identifies a session by the peer's TCP address and its AS number.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct PeerId {
    pub addr: SocketAddr,
    pub asn: u32,
}

impl std::fmt::Display for PeerId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}(AS{})", socket_text(self.addr), self.asn)
    }
}

/// Where the dialer is in bringing up a configured peer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DialState {
    /// Not dialling: the peer connected to us, or the dialer has not started.
    Idle,
    Connecting,
    /// The last attempt failed or the session ended; waiting to retry.
    Backoff {
        until: SystemTime,
        /// How the last attempt or session ended.
        last_error: String,
    },
    /// The peer is passive: not dialled, waiting for it to connect.
    Passive {
        /// How its last session, or the last connection from it, ended,
        /// if one did.
        last_error: Option<String>,
    },
}

/// A peer from the configuration file, whether or not it is currently up.
#[derive(Debug, Clone)]
pub struct ConfiguredPeer {
    pub name: String,
    pub endpoint: Endpoint,
    pub remote_asn: u32,
    /// What the peer is to us, if that was said.
    pub role: Option<PeerRole>,
    /// Whether it is left to the peer to connect.
    pub passive: bool,
    /// About how many routes it sends when all is well, if that was said.
    pub expected_routes: Option<u32>,
}

/// Counts kept by a session as it goes, for whoever wants to look.
#[derive(Debug, Default)]
pub struct SessionCounters {
    /// Routes of the peer's that are in the routing table now.
    routes: AtomicU64,
    /// Routes the peer announced that its import policy refused, since the
    /// session came up.
    rejected: AtomicU64,
    /// Routes of the peer's that are held out of the table as invalid, to
    /// be taken if route origin validation comes to say otherwise.
    held: AtomicU64,
    /// Routes of the peer's kept from its last session, which it has not
    /// sent again yet (graceful restart).
    stale: AtomicU64,
    /// Routes announced to the peer and not withdrawn since: what it
    /// holds from us, every route counted where it is sent several for
    /// a prefix.
    advertised: AtomicU64,
}

impl SessionCounters {
    pub fn routes(&self) -> u64 {
        self.routes.load(Ordering::Relaxed)
    }

    pub fn rejected(&self) -> u64 {
        self.rejected.load(Ordering::Relaxed)
    }

    pub fn held(&self) -> u64 {
        self.held.load(Ordering::Relaxed)
    }

    pub(crate) fn set_held(&self, held: usize) {
        self.held.store(held as u64, Ordering::Relaxed);
    }

    pub fn advertised(&self) -> u64 {
        self.advertised.load(Ordering::Relaxed)
    }

    pub(crate) fn set_advertised(&self, advertised: usize) {
        self.advertised.store(advertised as u64, Ordering::Relaxed);
    }

    pub fn stale(&self) -> u64 {
        self.stale.load(Ordering::Relaxed)
    }

    pub(crate) fn set_stale(&self, stale: usize) {
        self.stale.store(stale as u64, Ordering::Relaxed);
    }

    /// Notes that `cleared` stale routes were sent again or went.
    pub(crate) fn clear_stale(&self, cleared: usize) {
        // Only the session writes this.
        let left = self.stale().saturating_sub(cleared as u64);
        self.stale.store(left, Ordering::Relaxed);
    }

    /// Notes what one UPDATE did, and returns how many routes the peer has
    /// in the table after it.
    pub(crate) fn note(&self, added: usize, removed: usize, rejected: usize) -> u64 {
        self.rejected.fetch_add(rejected as u64, Ordering::Relaxed);
        // Only the session writes this, so the two steps are not a race.
        let routes = (self.routes() + added as u64).saturating_sub(removed as u64);
        self.routes.store(routes, Ordering::Relaxed);
        routes
    }
}

/// Handle to one established session. Cloning is cheap.
#[derive(Debug, Clone)]
pub struct SessionHandle {
    pub peer: PeerId,
    /// Name from the configuration, if this peer is configured.
    pub name: Option<String>,
    /// Which side opened the TCP connection.
    pub direction: Direction,
    /// Our side of the TCP connection; used as the next hop for routes we
    /// originate to this peer.
    pub local_addr: SocketAddr,
    pub established_at: SystemTime,
    /// Whether the peer is in our own AS.
    pub ibgp: bool,
    /// What the peer is to us, if that is known.
    pub role: Option<PeerRole>,
    /// Whether the peer said so itself in its OPEN (RFC 9234).
    pub role_confirmed: bool,
    /// Whether IPv4 routes go with IPv6 next hops on this session
    /// (RFC 8950).
    pub extended_next_hop: bool,
    /// Whether the peer negotiated RFC 6793 four-octet ASNs. Determines how
    /// AS_PATH is encoded on the wire for this session.
    pub four_octet_asn: bool,
    /// The address families negotiated with the peer. Routes of any other
    /// family are neither sent to it nor accepted from it.
    pub families: FamilySet,
    /// The families for which the peer sends us several routes for a
    /// prefix (ADD-PATH, RFC 7911), and those for which we send it path
    /// identifiers.
    pub add_path_receive: FamilySet,
    pub add_path_send: FamilySet,
    /// The peer's number in the routing table: what its routes there are
    /// marked with.
    pub rib_peer: PeerIndex,
    /// Graceful restart (RFC 4724), if both sides offered it.
    pub graceful_restart: Option<Agreed>,
    /// How many routes the peer has, and how many were refused.
    pub counters: Arc<SessionCounters>,
    /// What the peer is told, and how.
    pub export: Arc<ExportPolicy>,
    /// Limits on what the peer is told, where there are any for it.
    pub export_limits: Option<Arc<ExportLimits>>,
    /// The groups the session is in, for control communities.
    pub groups: Arc<[u32]>,
    /// Whether the peer is a route reflector client of ours (RFC 4456).
    pub reflector_client: bool,
    /// The peer's BGP Identifier.
    pub peer_router_id: Ipv4Addr,
    outbound: mpsc::Sender<Outbound>,
    /// Where the session is asked to end, if it listens for that.
    ending: Option<mpsc::UnboundedSender<SessionError>>,
    /// Which of the RPKI's checks the peer's routes are subject to, if
    /// any, and where the session is told to look at its routes again.
    validation: Option<(Option<Subject>, mpsc::UnboundedSender<Vec<Nlri>>)>,
}

#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error("session with {0} has closed")]
pub struct SessionClosed(pub PeerId);

impl SessionHandle {
    pub(crate) fn new(
        peer: PeerId,
        name: Option<String>,
        direction: Direction,
        local_addr: SocketAddr,
        negotiated: &Negotiated,
        rib_peer: PeerIndex,
        outbound: mpsc::Sender<Outbound>,
    ) -> Self {
        Self {
            peer,
            name,
            direction,
            local_addr,
            established_at: SystemTime::now(),
            ibgp: negotiated.ibgp,
            role: None,
            role_confirmed: false,
            extended_next_hop: negotiated.extended_next_hop,
            four_octet_asn: negotiated.four_octet_asn,
            families: negotiated.families.clone(),
            add_path_receive: negotiated.add_path_receive.clone(),
            add_path_send: negotiated.add_path_send.clone(),
            rib_peer,
            graceful_restart: negotiated.graceful_restart.clone(),
            counters: Arc::default(),
            export: Arc::default(),
            export_limits: None,
            groups: Arc::default(),
            reflector_client: false,
            peer_router_id: negotiated.peer_identifier,
            outbound,
            ending: None,
            validation: None,
        }
    }

    /// Has the session listen for being asked to end: what is asked with
    /// [`SessionHandle::end`] arrives at the receiver returned.
    pub(crate) fn listen_for_end(&mut self) -> mpsc::UnboundedReceiver<SessionError> {
        let (tx, rx) = mpsc::unbounded_channel();
        self.ending = Some(tx);
        rx
    }

    /// Asks the session to end because of `error`: it tells the peer with
    /// the NOTIFICATION for it, and closes. What was queued before may go
    /// out first; the caller queues nothing more.
    pub(crate) fn end(&self, error: SessionError) {
        if let Some(ending) = &self.ending {
            // Fails only if the session has ended already.
            let _ = ending.send(error);
        }
    }

    /// Marks the session's routes as subject to validation with the
    /// RPKI's data. What is sent with [`SessionHandle::revalidate`]
    /// arrives at the receiver returned.
    pub(crate) fn validate(
        &mut self,
        subject: Option<Subject>,
    ) -> mpsc::UnboundedReceiver<Vec<Nlri>> {
        let (tx, rx) = mpsc::unbounded_channel();
        self.validation = Some((subject, tx));
        rx
    }

    /// Which of the RPKI's checks the session's routes are subject to, if
    /// any.
    pub fn validated(&self) -> Option<Subject> {
        self.validation.as_ref().and_then(|(subject, _)| *subject)
    }

    /// Tells the session that the validation data has changed. `suspects`
    /// are routes of its peer's in the table that are now invalid.
    pub(crate) fn revalidate(&self, suspects: Vec<Nlri>) {
        if let Some((_, revalidate)) = &self.validation {
            // Fails only if the session has just ended.
            let _ = revalidate.send(suspects);
        }
    }

    /// What the peer goes by in the metrics: its configured name, or its
    /// address.
    pub fn metric_label(&self) -> String {
        self.name
            .clone()
            .unwrap_or_else(|| crate::address::PeerAddress::from(self.peer.addr).to_string())
    }

    /// Whether `other` is a handle to this same session, and not to another
    /// session with the same peer.
    pub fn is_same_session(&self, other: &SessionHandle) -> bool {
        self.outbound.same_channel(&other.outbound)
    }

    /// Queues a message for transmission on this session. Waits for queue
    /// space if the session's writer is behind.
    pub async fn send(&self, msg: Message) -> Result<(), SessionClosed> {
        self.outbound
            .send(Outbound::Message(msg))
            .await
            .map_err(|_| SessionClosed(self.peer))
    }

    /// Sends the peer a piece of one or more UPDATEs that are already
    /// encoded, headers and all, as the exporter builds them.
    pub async fn send_encoded(&self, piece: BytesMut) -> Result<(), SessionClosed> {
        self.outbound
            .send(Outbound::Encoded(piece))
            .await
            .map_err(|_| SessionClosed(self.peer))
    }
}

/// What the rest of the server gives a session to send: a message, or a
/// piece of one or more UPDATEs already encoded, headers and all, which
/// goes to the peer as it is once the session is established.
#[derive(Debug)]
pub enum Outbound {
    Message(Message),
    Encoded(BytesMut),
}

#[derive(Debug, Clone)]
pub enum SessionEvent {
    Up(SessionHandle),
    Down(PeerId),
    /// The peer sent ROUTE-REFRESH (RFC 2918): it wants everything we
    /// advertise to it sent again, typically because its import policy
    /// changed.
    RefreshRequested(SessionHandle),
}

#[derive(Clone)]
pub struct SessionRegistry {
    inner: Arc<RwLock<HashMap<PeerId, SessionHandle>>>,
    configured: Arc<Vec<ConfiguredPeer>>,
    dial_states: Arc<RwLock<HashMap<String, DialState>>>,
    shutting_down: Arc<AtomicBool>,
    /// How the daemon is leaving, once it is: every connection hears it.
    leaving: watch::Sender<Option<Leave>>,
    events: broadcast::Sender<SessionEvent>,
}

impl Default for SessionRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl SessionRegistry {
    pub fn new() -> Self {
        Self::with_configured(vec![])
    }

    pub fn with_configured(configured: Vec<ConfiguredPeer>) -> Self {
        let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
        let dial_states = configured
            .iter()
            .map(|p| {
                let state = if p.passive {
                    DialState::Passive { last_error: None }
                } else {
                    DialState::Idle
                };
                (p.name.clone(), state)
            })
            .collect();
        for peer in &configured {
            PEER_UP.with_label_values(&[&peer.name]).set(0);
        }
        Self {
            inner: Arc::new(RwLock::new(HashMap::new())),
            configured: Arc::new(configured),
            dial_states: Arc::new(RwLock::new(dial_states)),
            shutting_down: Arc::new(AtomicBool::new(false)),
            leaving: watch::Sender::new(None),
            events,
        }
    }

    /// Peers from the configuration file, each with the dialer's current
    /// state for it.
    pub fn configured_peers(&self) -> Vec<(ConfiguredPeer, DialState)> {
        let states = self.dial_states.reading();
        self.configured
            .iter()
            .map(|p| {
                (
                    p.clone(),
                    states.get(&p.name).cloned().unwrap_or(DialState::Idle),
                )
            })
            .collect()
    }

    pub(crate) fn set_dial_state(&self, name: &str, state: DialState) {
        self.dial_states.writing().insert(name.to_string(), state);
    }

    /// An established session with the peer at this address and with this
    /// ASN, regardless of which side opened the TCP connection.
    pub fn find(&self, address: PeerAddress, asn: u32) -> Option<SessionHandle> {
        self.inner
            .reading()
            .values()
            .find(|h| PeerAddress::from(h.peer.addr) == address && h.peer.asn == asn)
            .cloned()
    }

    /// Adds a session and emits [`SessionEvent::Up`]. The returned guard
    /// removes the session again when dropped.
    pub(crate) fn register(&self, handle: SessionHandle) -> SessionGuard {
        let peer = handle.peer;
        self.inner.writing().insert(peer, handle.clone());
        SESSIONS_ESTABLISHED.inc();
        if let Some(name) = &handle.name {
            PEER_UP.with_label_values(&[name]).set(1);
        }
        // Fails only when nobody is listening, which is fine.
        let _ = self.events.send(SessionEvent::Up(handle));
        SessionGuard {
            registry: self.clone(),
            peer,
        }
    }

    /// All currently established sessions, in no particular order.
    pub fn list(&self) -> Vec<SessionHandle> {
        self.inner.reading().values().cloned().collect()
    }

    /// Called by a session when its peer asks for a route refresh.
    pub(crate) fn request_refresh(&self, peer: &PeerId) {
        let handle = self.inner.reading().get(peer).cloned();
        if let Some(handle) = handle {
            let _ = self.events.send(SessionEvent::RefreshRequested(handle));
        }
    }

    /// True once [`leave`](Self::leave) has been called; dialers stop
    /// reconnecting.
    pub fn is_shutting_down(&self) -> bool {
        self.shutting_down.load(Ordering::Relaxed)
    }

    /// Closes every connection because the daemon is going, as `leave`
    /// says: for good, or to restart (see [`Leave`]). Waits up to `grace`
    /// for the sessions to finish; returns how many were still open.
    pub async fn leave(&self, leave: Leave, grace: Duration) -> usize {
        self.shutting_down.store(true, Ordering::Relaxed);
        // Every connection watches this, whatever state it is in; one whose
        // peer is not reading is not held up by the others.
        self.leaving.send_replace(Some(leave));
        let deadline = tokio::time::Instant::now() + grace;
        while !self.list().is_empty() && tokio::time::Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        self.list().len()
    }

    /// How the daemon is leaving, once it is.
    pub fn leaving(&self) -> watch::Receiver<Option<Leave>> {
        self.leaving.subscribe()
    }

    pub fn subscribe(&self) -> broadcast::Receiver<SessionEvent> {
        self.events.subscribe()
    }

    fn remove(&self, peer: &PeerId) {
        let removed = self.inner.writing().remove(peer);
        if let Some(handle) = &removed {
            SESSIONS_ESTABLISHED.dec();
            if let Some(name) = &handle.name {
                PEER_UP.with_label_values(&[name]).set(0);
            }
            let _ = self.events.send(SessionEvent::Down(*peer));
        }
    }
}

/// Removes the session from its registry on drop.
pub struct SessionGuard {
    registry: SessionRegistry,
    peer: PeerId,
}

impl Drop for SessionGuard {
    fn drop(&mut self) {
        self.registry.remove(&self.peer);
    }
}