ferrix/crates/web/src/events.rs

//! Server-sent event stream merging session, local-route and best-route
//! changes into one feed the UI can follow.

use std::convert::Infallible;
use std::sync::Arc;

use axum::extract::State;
use axum::response::sse::{Event, KeepAlive, Sse};
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rib::LocalRibEvent;
use bgp_rib::table::{self, Route};
use bgp_server::session::SessionEvent;
use futures::stream::{Stream, StreamExt};
use serde::Serialize;
use tokio::sync::mpsc;
use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
use tokio_stream::wrappers::{BroadcastStream, ReceiverStream};

use crate::AppState;
use crate::api::{LocalRouteView, PeerNames, PeerView, ReceivedRouteView};

#[derive(Debug, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum WireEvent {
    SessionUp {
        peer: PeerView,
    },
    SessionDown {
        peer_addr: String,
        peer_asn: u32,
    },
    /// The peer asked us to send all our routes again (ROUTE-REFRESH).
    RouteRefreshRequested {
        peer: PeerView,
    },
    LocalAnnounced {
        route: LocalRouteView,
    },
    LocalWithdrawn {
        prefix: String,
    },
    /// The best route for a prefix is now this one.
    BestChanged {
        route: ReceivedRouteView,
    },
    /// A prefix has no route any more.
    BestWithdrawn {
        prefix: String,
    },
    /// The client fell behind and missed events on `stream` (`missed` of
    /// them, where that is known); it should refetch the corresponding
    /// table.
    Lagged {
        stream: &'static str,
        missed: u64,
    },
}

impl WireEvent {
    fn from_session(ev: Result<SessionEvent, BroadcastStreamRecvError>) -> Self {
        match ev {
            Ok(SessionEvent::Up(h)) => Self::SessionUp {
                peer: PeerView::from(&h),
            },
            Ok(SessionEvent::Down(p)) => Self::SessionDown {
                peer_addr: bgp_server::address::socket_text(p.addr),
                peer_asn: p.asn,
            },
            Ok(SessionEvent::RefreshRequested(h)) => Self::RouteRefreshRequested {
                peer: PeerView::from(&h),
            },
            Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged {
                stream: "sessions",
                missed: n,
            },
        }
    }

    fn from_local(ev: Result<LocalRibEvent, BroadcastStreamRecvError>) -> Self {
        match ev {
            Ok(LocalRibEvent::Announced(r)) => Self::LocalAnnounced {
                route: LocalRouteView::from(&r),
            },
            Ok(LocalRibEvent::Withdrawn { prefix }) => Self::LocalWithdrawn {
                prefix: prefix.to_string(),
            },
            Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged {
                stream: "local_routes",
                missed: n,
            },
        }
    }
}

/// The table's changes of best route as events, from when the client
/// connects. Each client follows the table's journal with a reader of its
/// own, so one that reads slowly holds nobody else up.
fn best_routes(state: &AppState) -> impl Stream<Item = WireEvent> + use<> {
    let (events, stream) = mpsc::channel(BEST_ROUTE_BUFFER);
    let state_for_views = state.clone();
    // Taken before this returns, so that nothing that happens after the
    // client is answered is missed.
    let mut reader = state.rib.reader_from_now();
    tokio::spawn(async move {
        // If the reader falls too far behind it is given the whole table
        // again. That is not news: the client is told it missed something
        // and gets its table from the API.
        let mut syncing = false;
        loop {
            tokio::select! {
                () = reader.changed() => {}
                () = events.closed() => return,
            }
            loop {
                // Only what is needed is taken while the table is held;
                // the views are made afterwards.
                let mut changes = Vec::new();
                let progress = reader.poll(BEST_ROUTE_BUFFER, |event| match event {
                    table::Event::Reset => {
                        syncing = true;
                        changes.push(Change::Lagged);
                    }
                    table::Event::Synced => syncing = false,
                    table::Event::Best {
                        prefix,
                        peer,
                        path_id,
                        attrs,
                    } if !syncing => {
                        changes.push(Change::Best(Route {
                            prefix,
                            peer,
                            path_id,
                            attrs: Arc::clone(attrs),
                            best: true,
                            // Not said in the journal; the event stream
                            // does not show it.
                            stale: false,
                        }));
                    }
                    table::Event::Withdrawn { prefix } if !syncing => {
                        changes.push(Change::Withdrawn(prefix));
                    }
                    _ => {}
                });
                let names = PeerNames::of(&state_for_views);
                for change in changes {
                    let event = match change {
                        Change::Best(route) => match names.view(&route) {
                            Some(route) => WireEvent::BestChanged { route },
                            None => continue,
                        },
                        Change::Withdrawn(prefix) => WireEvent::BestWithdrawn {
                            prefix: prefix.to_string(),
                        },
                        Change::Lagged => WireEvent::Lagged {
                            stream: "best_routes",
                            missed: 0,
                        },
                    };
                    if events.send(event).await.is_err() {
                        return;
                    }
                }
                if progress.caught_up {
                    break;
                }
            }
        }
    });
    ReceiverStream::new(stream)
}

/// How many best-route events are read from the journal, and queued for a
/// client, at a time.
const BEST_ROUTE_BUFFER: usize = 256;

enum Change {
    Best(Route),
    Withdrawn(IpPrefix),
    /// The client's reader fell too far behind the table.
    Lagged,
}

pub async fn stream(
    State(state): State<AppState>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let sessions = BroadcastStream::new(state.registry.subscribe()).map(WireEvent::from_session);
    let local = BroadcastStream::new(state.local_rib.subscribe()).map(WireEvent::from_local);
    let received = best_routes(&state);

    let merged =
        futures::stream::select(sessions, futures::stream::select(local, received)).map(|ev| {
            Ok(Event::default()
                .event("bgp")
                .json_data(&ev)
                .unwrap_or_else(|e| Event::default().event("error").data(e.to_string())))
        });

    Sse::new(merged).keep_alive(KeepAlive::default())
}