ferrix/crates/server/src/advertise.rs

//! Announces routes to every established session: those this router
//! originates, and, to the sessions whose export policy says so, the best
//! of those learned from others.
//!
//! Each session has an exporter of its own: a task that sends the session
//! the originated routes and then every change to them. An exporter waits
//! when its session's queue is full, which is what a peer that reads slowly
//! or not at all comes to, and it is the only thing that waits: the other
//! sessions have their own exporters and are not held up.
//!
//! An exporter remembers what it has announced. When it has waited so long
//! that it missed changes, it compares that with the routes as they now
//! are, announces those, and withdraws what is no longer among them. A peer
//! that was slow ends up with what every other peer has.
//!
//! The advertiser itself only starts an exporter when a session comes up
//! and stops it when the session goes.
//!
//! To a peer above or beside us, the export limits are a check on all of
//! this ([`crate::policy::ExportLimits`]): routes for prefixes outside them are not
//! announced, whatever else says they are to be, and if more routes were
//! to be announced than they allow, none of them is and the session ends.
//!
//! To a peer that agreed to graceful restart (RFC 4724) an exporter sends
//! an End-of-RIB marker for each address family once the peer has all our
//! routes: at once after the originated ones, or, where learned routes are
//! passed on too, once the routing table has been gone through and every
//! peer has sent its own routes since the start. A peer that kept our
//! routes over a restart of ours drops those it has not had again then.

use std::collections::{HashMap, HashSet};
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::sync::Arc;
use std::time::Duration;

use bgp_packet::message::{
    AsPathAttribute, AsPathSegment, CommunitiesPathAttribute, LargeCommunitiesPathAttribute,
    LocalPrefPathAttribute, Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute,
    MultiExitDiscPathAttribute, NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage,
};
use bgp_primitives::constants::{AfiSafi, IpVersion};
use bgp_primitives::ip_prefix::{IpPrefix, Nlri};
use bgp_primitives::parser::ParserContext;
use bgp_rib::table::{AttrSet, DEFAULT_LOCAL_PREF, Event, PeerIndex, Reader, RibTable};
use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute};
use bytes::BytesMut;
use prometheus::IntGauge;
use tokio::sync::broadcast::Receiver;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::{Notify, OwnedSemaphorePermit, Semaphore, watch};
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};

use crate::error::SessionError;
use crate::export::{self, Egress, Outgoing, Reflection, Source};
use crate::family::FamilySet;
use crate::marks::Control;
use crate::metrics::{PEER_ADVERTISED, ROUTES_EXPORTED};
use crate::policy::PeerRole;
use crate::session::{PeerId, SessionClosed, SessionEvent, SessionHandle, SessionRegistry};

/// Starts the advertiser. It hears of every session that comes up after
/// this returns, so call it before anything that can bring one up.
pub fn spawn_advertiser(
    local_rib: LocalRib,
    rib: Arc<RibTable>,
    registry: SessionRegistry,
    local_asn: u32,
    control: Option<Control>,
    cluster_id: Ipv4Addr,
    converged: watch::Receiver<bool>,
) -> JoinHandle<()> {
    let advertiser = Advertiser {
        converged,
        session_events: registry.subscribe(),
        local_rib,
        rib,
        registry,
        local_asn,
        control,
        cluster_id,
        resyncs: Arc::new(Semaphore::new(RESYNCS_AT_ONCE)),
        exporters: HashMap::new(),
    };
    tokio::spawn(advertiser.run())
}

/// An exporter, as the advertiser holds it.
struct Running {
    session: SessionHandle,
    /// Tells the exporter to send everything again.
    resend: Arc<Notify>,
    task: JoinHandle<()>,
}

impl Drop for Running {
    fn drop(&mut self) {
        self.task.abort();
    }
}

struct Advertiser {
    local_rib: LocalRib,
    /// The routes learned from peers, of which the best are passed on.
    rib: Arc<RibTable>,
    registry: SessionRegistry,
    local_asn: u32,
    control: Option<Control>,
    /// The cluster this router reflects routes for (RFC 4456).
    cluster_id: Ipv4Addr,
    /// Leave for a session to gather the whole table before it sends it;
    /// see `Learned::gathered`.
    resyncs: Arc<Semaphore>,
    session_events: Receiver<SessionEvent>,
    exporters: HashMap<PeerId, Running>,
    /// Whether every peer has sent its routes since the start.
    converged: watch::Receiver<bool>,
}

impl Advertiser {
    /// Runs for as long as the server does. The server watches for this
    /// task ending: without it nothing is announced.
    async fn run(mut self) {
        loop {
            match self.session_events.recv().await {
                Ok(SessionEvent::Up(session)) => self.start(session),
                Ok(SessionEvent::Down(peer)) => {
                    self.exporters.remove(&peer);
                }
                Ok(SessionEvent::RefreshRequested(session)) => {
                    if let Some(exporter) = self.exporters.get(&session.peer) {
                        debug!(peer = %session.peer, "Route refresh requested, re-sending our routes");
                        exporter.resend.notify_one();
                    }
                }
                Err(RecvError::Lagged(missed)) => {
                    warn!(
                        missed,
                        "Advertiser lagged on session events, checking every session"
                    );
                    self.reconcile();
                }
                // The registry is gone, and the server with it.
                Err(RecvError::Closed) => return,
            }
        }
    }

    /// Starts an exporter for a session that has come up.
    fn start(&mut self, session: SessionHandle) {
        let resend = Arc::new(Notify::new());
        let exporter = Exporter {
            // Subscribed here, before the exporter first looks at the
            // routes, so that no change between the two is missed.
            changes: self.local_rib.subscribe(),
            local_rib: self.local_rib.clone(),
            // A follower of the best routes, if the peer is to be told
            // of them, or of every route if it is to have them all and
            // agreed to tell them apart: it is given the whole table first.
            learned: session.export.learned_routes.then(|| {
                let all_paths = session.export.all_paths && !session.add_path_send.is_empty();
                Learned::new(self.learned_reader(all_paths), all_paths)
            }),
            rib: Arc::clone(&self.rib),
            resyncs: Arc::clone(&self.resyncs),
            control: self.control,
            cluster_id: self.cluster_id,
            session: session.clone(),
            local_asn: self.local_asn,
            resend: Arc::clone(&resend),
            own_announced: HashSet::new(),
            end_of_rib: session
                .graceful_restart
                .as_ref()
                .map(|_| session.families.clone()),
            converged: Some(self.converged.clone()).filter(|converged| !*converged.borrow()),
            advertised_metric: PEER_ADVERTISED.with_label_values(&[&session.metric_label()]),
        };
        let running = Running {
            session: session.clone(),
            resend,
            task: tokio::spawn(exporter.run()),
        };
        // A session with the same peer that this one replaces has gone,
        // whether or not we have heard so yet; its exporter goes with it.
        self.exporters.insert(session.peer, running);
    }

    /// A new follower of the table for an exporter: of every route, or of
    /// the best ones.
    fn learned_reader(&self, all_paths: bool) -> Reader {
        if all_paths {
            self.rib.reader_of_all_routes()
        } else {
            self.rib.reader()
        }
    }

    /// Makes the exporters match the sessions there are, after events were
    /// missed: sessions may have come and gone unheard, and peers may have
    /// asked for their routes again.
    fn reconcile(&mut self) {
        let sessions = self.registry.list();
        self.exporters.retain(|_, exporter| {
            sessions
                .iter()
                .any(|session| session.is_same_session(&exporter.session))
        });
        for session in sessions {
            match self.exporters.get(&session.peer) {
                Some(exporter) => exporter.resend.notify_one(),
                None => self.start(session),
            }
        }
    }
}

/// Keeps one session's peer up to date with the routes we originate and,
/// if it is to be told of them, the best of those learned from others.
struct Exporter {
    rib: Arc<RibTable>,
    resyncs: Arc<Semaphore>,
    control: Option<Control>,
    cluster_id: Ipv4Addr,
    learned: Option<Learned>,
    local_rib: LocalRib,
    changes: Receiver<LocalRibEvent>,
    session: SessionHandle,
    local_asn: u32,
    resend: Arc<Notify>,
    /// The prefixes of our own routes announced to the peer and not
    /// withdrawn since: our Adj-RIB-Out for it (RFC 4271 §3.2), as far as
    /// it needs keeping. The learned ones are in `learned`.
    own_announced: HashSet<IpPrefix>,
    /// The families to send End-of-RIB markers for, until they are sent;
    /// `None` if graceful restart was not agreed.
    end_of_rib: Option<FamilySet>,
    /// Word of every peer having sent its routes since the start, until it
    /// comes.
    converged: Option<watch::Receiver<bool>>,
    /// How many routes the peer holds from us, for whoever asks.
    advertised_metric: IntGauge,
}

/// How many changes of best route are taken from the table at a time.
const EXPORT_BATCH: usize = 65_536;

/// How long changes are gathered before they go out, once there are any.
const COALESCE: Duration = Duration::from_millis(50);

/// How many sessions may hold a whole table in hand at once while it is
/// given out to them anew: some tens of megabytes each for the Internet.
const RESYNCS_AT_ONCE: usize = 4;

/// How many routes ready to go out are remembered, by the attributes they
/// came with. A table has a few hundred thousand distinct sets, each a
/// kilobyte or so in this form; what a flush is sending holds its own.
const DRESSED_KEPT: usize = 65_536;

fn version_index(version: IpVersion) -> usize {
    match version {
        IpVersion::Ipv4 => 0,
        IpVersion::Ipv6 => 1,
    }
}

/// A change of best route, as taken out of the table to be dealt with
/// once the table is let go of.
enum News {
    Reset,
    Synced,
    Best {
        prefix: IpPrefix,
        peer: PeerIndex,
        attrs: Arc<AttrSet>,
    },
    Withdrawn(IpPrefix),
    /// All the routes of `prefix`, for a session that is told of every
    /// route: who each is from, that peer's identifier for it, its
    /// attributes, and whether it is the best.
    Routes {
        prefix: IpPrefix,
        routes: Vec<(PeerIndex, u32, Arc<AttrSet>, bool)>,
    },
}

/// What a learned route goes out as depends on: the attributes it came
/// with (by where they are), who it came from, and its IP version.
type DressedKey = (usize, Source, IpVersion);

/// A learned route as this session announces it.
struct Dressed {
    outgoing: Outgoing,
    /// How many octets its attributes take on this session's wire.
    octets: usize,
}

/// A route of a prefix as a session that sends every route tells it
/// apart: the peer it is from, and that peer's identifier for it.
type PathKey = (PeerIndex, u32);

/// A route announced to a peer that is sent every route, and not
/// withdrawn since.
struct Sent {
    key: PathKey,
    /// The path identifier it went out with (ADD-PATH, RFC 7911), which
    /// stays its own for as long as it is announced.
    id: u32,
    /// The attributes it went out with, as the table has them: what it
    /// goes out as follows from those, so a route whose set is the same
    /// has not changed. The table holds the set anyway; the form it goes
    /// out in is a kilobyte a set and is not kept here.
    attrs: Arc<AttrSet>,
}

/// What the peer is to end up with for a prefix.
enum Verdict {
    /// The best route as it goes out, or `None` for a prefix the peer is
    /// not to have.
    Best(Option<Arc<Dressed>>),
    /// Every route that goes out, with what tells it apart and the
    /// attributes it came with; empty for a prefix the peer is not to
    /// have.
    All(Vec<(PathKey, Arc<AttrSet>, Arc<Dressed>)>),
}

/// What the peer is to end up with for each prefix heard of.
type Verdicts = HashMap<IpPrefix, Verdict>;

/// The learned routes announced to the peer: our Adj-RIB-Out for it, as
/// far as it needs keeping.
enum Announced {
    /// The prefixes, each with its best route.
    Best(HashSet<IpPrefix>),
    /// For each prefix, every route announced, and how many in all.
    All {
        routes: HashMap<IpPrefix, Vec<Sent>>,
        count: usize,
    },
}

impl Announced {
    fn new(all_paths: bool) -> Self {
        if all_paths {
            Self::All {
                routes: HashMap::new(),
                count: 0,
            }
        } else {
            Self::Best(HashSet::new())
        }
    }

    /// How many routes the peer holds.
    fn routes(&self) -> usize {
        match self {
            Self::Best(prefixes) => prefixes.len(),
            Self::All { count, .. } => *count,
        }
    }

    /// Takes everything out, leaving this empty, as what the peer holds
    /// in doubt until it is heard of again.
    fn take(&mut self) -> Self {
        match self {
            Self::Best(prefixes) => Self::Best(std::mem::take(prefixes)),
            Self::All { routes, count } => Self::All {
                routes: std::mem::take(routes),
                count: std::mem::take(count),
            },
        }
    }

    /// Adds what `other` holds.
    fn absorb(&mut self, other: Self) {
        match (self, other) {
            (Self::Best(prefixes), Self::Best(more)) => prefixes.extend(more),
            (Self::All { routes, count }, Self::All { routes: more, .. }) => {
                for (prefix, sent) in more {
                    *count += sent.len();
                    routes.insert(prefix, sent);
                }
            }
            // Both are of one session, which is one or the other.
            _ => {}
        }
    }

    /// Everything, as withdrawals, by IP version.
    fn withdrawals(self) -> [Vec<Nlri>; 2] {
        let mut withdrawn: [Vec<Nlri>; 2] = Default::default();
        match self {
            Self::Best(prefixes) => {
                for prefix in prefixes {
                    withdrawn[version_index(prefix.address_family())].push(prefix.into());
                }
            }
            Self::All { routes, .. } => {
                for (prefix, sent) in routes {
                    let withdrawn = &mut withdrawn[version_index(prefix.address_family())];
                    withdrawn.extend(sent.iter().map(|sent| Nlri {
                        prefix,
                        path_id: sent.id,
                    }));
                }
            }
        }
        withdrawn
    }
}

/// What an exporter knows of the learned routes it has passed on.
struct Learned {
    reader: Reader,
    /// The verdicts so far on a table that is being given out anew, kept
    /// until the table is all there; and the leave to keep them.
    gathered: Option<Verdicts>,
    permit: Option<OwnedSemaphorePermit>,
    /// Whether every route goes, and not only the best of each prefix:
    /// the reader is then one of all routes. Only for a session on which
    /// routes are told apart by path identifiers for at least one family;
    /// for a family they are not, the best goes all the same.
    all_paths: bool,
    /// What is announced to the peer and not withdrawn since.
    announced: Announced,
    /// Whether the peer has been told of the whole table once.
    synced: bool,
    /// While the table is being gone through anew: what was announced
    /// before and has not been heard of again yet.
    stale: Option<Announced>,
    /// Routes as they go out, by the attributes they came with, who they
    /// came from and their IP version: a table's worth of routes share a
    /// few sets of attributes, and each is worked out once. The key holds
    /// the attributes, so that it stays theirs.
    dressed: HashMap<DressedKey, (Arc<AttrSet>, Option<Arc<Dressed>>)>,
}

impl Learned {
    /// How many learned routes the peer holds from us: those announced,
    /// and those not heard of again yet since the table was reset.
    fn held_by_peer(&self) -> usize {
        self.announced.routes() + self.stale.as_ref().map_or(0, Announced::routes)
    }

    /// The verdict that the peer is to have nothing for a prefix.
    fn nothing(&self) -> Verdict {
        if self.all_paths {
            Verdict::All(Vec::new())
        } else {
            Verdict::Best(None)
        }
    }

    /// Takes the next changes of best route from the table: all of what
    /// there is, up to a limit, so that it goes out together. Only what is
    /// needed is taken while the table is held.
    fn drain(&mut self) -> Vec<News> {
        let mut news = Vec::new();
        while news.len() < EXPORT_BATCH {
            let progress = self.reader.poll(EXPORT_BATCH - news.len(), |event| {
                news.push(match event {
                    Event::Reset => News::Reset,
                    Event::Synced => News::Synced,
                    Event::Best {
                        prefix,
                        peer,
                        attrs,
                        ..
                    } => News::Best {
                        prefix,
                        peer,
                        attrs: Arc::clone(attrs),
                    },
                    Event::Withdrawn { prefix } => News::Withdrawn(prefix),
                    Event::Routes { prefix, routes } => News::Routes {
                        prefix,
                        routes: routes
                            .iter()
                            .map(|route| {
                                (
                                    route.peer,
                                    route.path_id,
                                    Arc::clone(route.attrs),
                                    route.best,
                                )
                            })
                            .collect(),
                    },
                });
            });
            if progress.caught_up || progress.events == 0 {
                break;
            }
        }
        news
    }

    fn new(reader: Reader, all_paths: bool) -> Self {
        Self {
            reader,
            gathered: None,
            permit: None,
            all_paths,
            announced: Announced::new(all_paths),
            synced: false,
            stale: None,
            dressed: HashMap::new(),
        }
    }

    /// The route with `attrs` from `source` as it goes out on `egress`
    /// for prefixes of IP version `version`; `None` if it does not.
    fn dressed(
        &mut self,
        attrs: &Arc<AttrSet>,
        source: Source,
        version: IpVersion,
        egress: &Egress<'_>,
    ) -> Option<Arc<Dressed>> {
        let key = (Arc::as_ptr(attrs) as usize, source, version);
        if let Some((_, dressed)) = self.dressed.get(&key) {
            return dressed.clone();
        }
        let dressed = attrs
            .attributes()
            .and_then(|(attributes, next_hop)| {
                egress.outgoing(attributes, next_hop, source, version)
            })
            .map(|outgoing| {
                // As long as they will be on the wire, to know how many
                // prefixes fit beside them.
                let ctx = ParserContext {
                    four_octet_asn: egress.towards.four_octet_asn,
                    ibgp: egress.towards.ibgp,
                    add_path: Vec::new(),
                };
                let mut wire = BytesMut::new();
                let mut attributes = outgoing.attributes.clone();
                export::finish(&mut attributes, &egress.towards);
                let octets = attributes
                    .iter()
                    .try_for_each(|attribute| attribute.to_wire(&ctx, &mut wire))
                    .map_or(usize::MAX, |()| wire.len());
                Arc::new(Dressed { outgoing, octets })
            })
            // One that cannot be written is one that does not go.
            .filter(|dressed| dressed.octets < 3072);
        if self.dressed.len() >= DRESSED_KEPT {
            self.dressed.clear();
        }
        self.dressed
            .insert(key, (Arc::clone(attrs), dressed.clone()));
        dressed
    }
}

/// Who a route from `peer` is from, as the export policy sees it, asked
/// of the table once for each peer and remembered in `sources`. `None` if
/// the peer has gone, and its routes with it.
fn source_of(
    rib: &RibTable,
    sources: &mut HashMap<PeerIndex, Option<Source>>,
    peer: PeerIndex,
) -> Option<Source> {
    *sources.entry(peer).or_insert_with(|| {
        rib.peer(peer).map(|info| Source {
            peer,
            ibgp: info.ibgp,
            reflector_client: info.reflector_client,
            router_id: info.router_id,
        })
    })
}

/// Waits until every peer has sent its routes since the start; for ever if
/// that is not to be waited for.
async fn converged(converged: &mut Option<watch::Receiver<bool>>) {
    match converged {
        // If the sender is gone, so is the server: nothing to wait for.
        Some(converged) => {
            let _ = converged.wait_for(|done| *done).await;
        }
        None => std::future::pending().await,
    }
}

/// Waits until there may be changes of best route to pass on; for ever if
/// the peer is not told of them.
async fn changed(learned: &mut Option<Learned>) {
    match learned {
        Some(learned) => learned.reader.changed().await,
        None => std::future::pending().await,
    }
}

/// Why an exporter has nothing more to do.
struct Finished;

impl From<SessionClosed> for Finished {
    fn from(_: SessionClosed) -> Self {
        Self
    }
}

impl Exporter {
    /// Runs until the session closes or the local RIB stops; the
    /// advertiser stops it when it hears the session has gone.
    async fn run(mut self) {
        info!(peer = %self.session.peer, "Session up, sending it our routes");
        let _ = self.export().await;
    }

    async fn export(&mut self) -> Result<(), Finished> {
        self.send_all().await?;
        self.end_of_rib().await?;
        loop {
            tokio::select! {
                change = self.changes.recv() => match change {
                    Ok(LocalRibEvent::Announced(route)) => self.announce(&route).await?,
                    Ok(LocalRibEvent::Withdrawn { prefix }) => self.withdraw(prefix).await?,
                    Err(RecvError::Lagged(missed)) => {
                        info!(
                            peer = %self.session.peer,
                            missed,
                            "Peer fell behind on local routes, bringing it up to date"
                        );
                        self.send_all().await?;
                    }
                    Err(RecvError::Closed) => return Err(Finished),
                },
                () = self.resend.notified() => {
                    self.send_all().await?;
                    // Everything, again: a new follower starts with the
                    // whole table.
                    if let Some(learned) = &mut self.learned {
                        learned.reader = if learned.all_paths {
                            self.rib.reader_of_all_routes()
                        } else {
                            self.rib.reader()
                        };
                    }
                }
                () = changed(&mut self.learned) => {
                    // Changes come in runs: a table arriving, a peer
                    // going. Gathered for a moment, the routes that go out
                    // alike go in one message, and the whole run costs one
                    // look at what each peer is told.
                    tokio::time::sleep(COALESCE).await;
                    self.follow().await?;
                    self.end_of_rib().await?;
                }
                () = converged(&mut self.converged), if self.converged.is_some() => {
                    self.converged = None;
                    self.end_of_rib().await?;
                }
            }
        }
    }

    /// Sends the End-of-RIB markers, if they are still to be sent and the
    /// peer has had all our routes: the originated ones, and the learned
    /// ones as far as there are to be any.
    async fn end_of_rib(&mut self) -> Result<(), Finished> {
        let complete = match &self.learned {
            None => true,
            Some(learned) => learned.synced && self.converged.is_none(),
        };
        if !complete {
            return Ok(());
        }
        let Some(families) = self.end_of_rib.take() else {
            return Ok(());
        };
        debug!(peer = %self.session.peer, %families, "Sending End-of-RIB");
        for family in families.iter() {
            self.session
                .send(Message::Update(UpdateMessage::end_of_rib(family)))
                .await?;
        }
        Ok(())
    }

    /// What every route on its way to `session` goes through. Not a method:
    /// `follow` holds the learned routes mutably while it uses this.
    fn egress(
        session: &SessionHandle,
        control: Option<Control>,
        cluster_id: Ipv4Addr,
        towards: Towards,
    ) -> Egress<'_> {
        Egress {
            towards,
            peer_asn: session.peer.asn,
            rib_peer: session.rib_peer,
            policy: &session.export,
            groups: &session.groups,
            control,
            reflection: Reflection {
                cluster_id,
                client: session.reflector_client,
                peer_router_id: session.peer_router_id,
            },
        }
    }

    /// Notes how many routes the peer now holds from us, for the status
    /// page and the metrics.
    fn note_advertised(&self) {
        let advertised =
            self.own_announced.len() + self.learned.as_ref().map_or(0, Learned::held_by_peer);
        self.session.counters.set_advertised(advertised);
        self.advertised_metric.set(advertised as i64);
    }

    /// The session as far as it decides what goes out on it and how.
    fn towards(&self) -> Towards {
        let session = &self.session;
        Towards {
            local_asn: self.local_asn,
            next_hop: session.local_addr.ip(),
            four_octet_asn: session.four_octet_asn,
            ibgp: session.ibgp,
            role: session.role,
            extended_next_hop: session.extended_next_hop,
        }
    }

    /// Takes the next changes of best route from the table and tells the
    /// peer of those that are for it.
    async fn follow(&mut self) -> Result<(), Finished> {
        let Some(learned) = &mut self.learned else {
            return Ok(());
        };
        let news = learned.drain();
        match self.judge(news).await {
            Some((verdicts, synced)) => self.flush(verdicts, synced).await,
            // Not all there yet.
            None => Ok(()),
        }
    }

    /// Decides what the peer is to end up with for each prefix the news
    /// is about: the last word on a prefix counts. While the whole table
    /// is being given out anew, which is shard by shard, the verdicts are
    /// kept until it is all there, and this returns `None`: the routes
    /// that share attributes are in every shard, and are only sent
    /// together if gathered first. Otherwise the verdicts, and whether
    /// the table has been given out whole.
    async fn judge(&mut self, news: Vec<News>) -> Option<(Verdicts, bool)> {
        let towards = self.towards();
        let session = &self.session;
        let learned = self.learned.as_mut()?;
        let limits = session.export_limits.as_deref();
        let egress = Self::egress(session, self.control, self.cluster_id, towards);
        let mut verdicts = learned.gathered.take().unwrap_or_default();
        let mut synced = false;
        // Who the routes are from, asked of the table once for each.
        let mut sources: HashMap<PeerIndex, Option<Source>> = HashMap::new();
        // Whether a learned route for `prefix` may go to the peer at all:
        // what we originate ourselves is announced as that, and not as
        // someone else's route for it.
        let may_go = |prefix: &IpPrefix| {
            session.families.carries(prefix)
                && !self.own_announced.contains(prefix)
                && limits.is_none_or(|limits| limits.allows(prefix))
        };
        for item in news {
            match item {
                News::Reset => {
                    if learned.synced {
                        // The table was changing faster than this exporter
                        // read it, and the journal let the oldest changes
                        // go: the whole table is gone through again, and
                        // what is not heard of again is withdrawn then.
                        info!(
                            peer = %session.peer,
                            "The table changed too fast to follow for the peer; going through it again"
                        );
                    }
                    // Whatever was announced and is not heard of again by
                    // the time the table is through is withdrawn then.
                    let known = learned.announced.take();
                    learned
                        .stale
                        .get_or_insert_with(|| Announced::new(learned.all_paths))
                        .absorb(known);
                    verdicts.clear();
                    synced = false;
                    // Gathering a table takes memory: only so many sessions
                    // do it at once, the others wait their turn.
                    if learned.permit.is_none() {
                        learned.permit = self.resyncs.clone().acquire_owned().await.ok();
                    }
                }
                News::Synced => {
                    synced = true;
                    learned.permit = None;
                }
                News::Withdrawn(prefix) => {
                    verdicts.insert(prefix, learned.nothing());
                }
                News::Best {
                    prefix,
                    peer,
                    attrs,
                } => {
                    let dressed = match source_of(&self.rib, &mut sources, peer) {
                        // What we originate ourselves is announced as
                        // that, and not as someone else's route for it.
                        Some(source) if may_go(&prefix) => {
                            learned.dressed(&attrs, source, prefix.address_family(), &egress)
                        }
                        // The peer has gone, and its routes with it.
                        _ => None,
                    };
                    verdicts.insert(prefix, Verdict::Best(dressed));
                }
                News::Routes { prefix, routes } => {
                    let version = prefix.address_family();
                    // Where the peer did not agree to tell the routes of a
                    // prefix apart, only the best goes, as plain BGP has it.
                    let told_apart = session.add_path_send.contains(AfiSafi::unicast(version));
                    let mut wanted = Vec::new();
                    if may_go(&prefix) {
                        for (peer, path_id, attrs, best) in routes {
                            if !told_apart && !best {
                                continue;
                            }
                            let Some(source) = source_of(&self.rib, &mut sources, peer) else {
                                continue;
                            };
                            if let Some(dressed) = learned.dressed(&attrs, source, version, &egress)
                            {
                                wanted.push(((peer, path_id), attrs, dressed));
                            }
                        }
                    }
                    verdicts.insert(prefix, Verdict::All(wanted));
                }
            }
        }
        if learned.permit.is_some() && !synced {
            learned.gathered = Some(verdicts);
            return None;
        }
        Some((verdicts, synced))
    }

    /// Sends the peer what the verdicts say: withdrawals first, then
    /// announcements, routes that go out alike together. `synced` says
    /// the table was given out whole, so what was not heard of again is
    /// withdrawn too.
    async fn flush(&mut self, verdicts: Verdicts, synced: bool) -> Result<(), Finished> {
        let towards = self.towards();
        let session = &self.session;
        let Some(learned) = &mut self.learned else {
            return Ok(());
        };
        let mut withdrawn: [Vec<Nlri>; 2] = Default::default();
        let mut batches: HashMap<(usize, IpVersion), (Arc<Dressed>, Vec<Nlri>)> = HashMap::new();
        for (prefix, verdict) in verdicts {
            let version = prefix.address_family();
            let withdrawn = &mut withdrawn[version_index(version)];
            let mut announce = |dressed: Arc<Dressed>, nlri: Nlri| {
                batches
                    .entry((Arc::as_ptr(&dressed) as usize, version))
                    .or_insert_with(|| (dressed, Vec::new()))
                    .1
                    .push(nlri);
            };
            match (verdict, &mut learned.announced, &mut learned.stale) {
                (Verdict::Best(verdict), Announced::Best(announced), stale) => {
                    // Heard of again: no longer in doubt, whichever way it
                    // goes.
                    let was_stale = match stale {
                        Some(Announced::Best(stale)) => stale.remove(&prefix),
                        _ => false,
                    };
                    match verdict {
                        Some(dressed) => {
                            announced.insert(prefix);
                            announce(dressed, prefix.into());
                        }
                        None => {
                            if announced.remove(&prefix) || was_stale {
                                withdrawn.push(prefix.into());
                            }
                        }
                    }
                }
                (Verdict::All(wanted), Announced::All { routes, count }, stale) => {
                    // What the peer has for the prefix: announced, or in
                    // doubt since the table was reset.
                    let mut had = routes.remove(&prefix).unwrap_or_default();
                    *count -= had.len();
                    if let Some(Announced::All { routes, count }) = stale
                        && let Some(stale) = routes.remove(&prefix)
                    {
                        *count -= stale.len();
                        had.extend(stale);
                    }
                    // A route it had keeps its identifier, and goes out
                    // again only if it changed; a new one gets the lowest
                    // identifier not in use for the prefix.
                    let mut has: Vec<Sent> = Vec::with_capacity(wanted.len());
                    for (key, attrs, dressed) in wanted {
                        let id = match had.iter().position(|sent| sent.key == key) {
                            Some(at) => {
                                let sent = had.swap_remove(at);
                                if Arc::ptr_eq(&sent.attrs, &attrs) {
                                    has.push(sent);
                                    continue;
                                }
                                sent.id
                            }
                            None => (1..)
                                .find(|id| !had.iter().chain(&has).any(|sent| sent.id == *id))
                                .expect("fewer routes than identifiers"),
                        };
                        announce(
                            Arc::clone(&dressed),
                            Nlri {
                                prefix,
                                path_id: id,
                            },
                        );
                        has.push(Sent { key, id, attrs });
                    }
                    withdrawn.extend(had.iter().map(|sent| Nlri {
                        prefix,
                        path_id: sent.id,
                    }));
                    *count += has.len();
                    if !has.is_empty() {
                        routes.insert(prefix, has);
                    }
                }
                // A verdict is of the kind the session's bookkeeping is.
                _ => {}
            }
        }
        if synced {
            learned.synced = true;
            if let Some(stale) = learned.stale.take() {
                for (version, routes) in withdrawn.iter_mut().zip(stale.withdrawals()) {
                    version.extend(routes);
                }
            }
        }
        // What the peer would hold once this goes out: none of it goes if
        // that is more than it may.
        let routes = self.own_announced.len() + learned.held_by_peer();
        let limits = session.export_limits.as_deref();
        if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) {
            return Err(over_limit(session, limit, routes));
        }
        // Counted as it goes out, not after: a whole table takes a while
        // to reach a peer that reads it at the pace of the Internet.
        let note = |announced: usize| {
            self.session.counters.set_advertised(announced);
            self.advertised_metric.set(announced as i64);
        };
        for (version, routes) in [IpVersion::Ipv4, IpVersion::Ipv6]
            .into_iter()
            .zip(withdrawn)
        {
            if routes.is_empty() {
                continue;
            }
            ROUTES_EXPORTED
                .with_label_values(&["withdrawn"])
                .inc_by(routes.len() as u64);
            for update in export::withdrawals(version, &routes, &towards) {
                session.send(Message::Update(update)).await?;
            }
        }
        let before = routes
            - batches
                .values()
                .map(|(_, routes)| routes.len())
                .sum::<usize>();
        let mut announced = 0;
        for ((_, version), (dressed, routes)) in batches {
            ROUTES_EXPORTED
                .with_label_values(&["announced"])
                .inc_by(routes.len() as u64);
            let updates = export::announcements(
                &dressed.outgoing,
                version,
                &routes,
                &towards,
                dressed.octets,
            );
            for update in updates {
                session.send(Message::Update(update)).await?;
            }
            announced += routes.len();
            note(before + announced);
        }
        self.note_advertised();
        Ok(())
    }

    /// Brings the peer to the routes as they now are: announces every one
    /// of them, and withdraws what was announced and is no longer there.
    ///
    /// Changes made while this runs are in `changes`, to be handled after.
    /// Some of those may be older than the routes read here; handling them
    /// again, in order, ends in the same place.
    async fn send_all(&mut self) -> Result<(), Finished> {
        let routes = self.local_rib.snapshot().await.map_err(|_| Finished)?;
        let current: HashSet<IpPrefix> = routes.iter().map(|route| route.prefix).collect();
        let gone: Vec<IpPrefix> = self.own_announced.difference(&current).copied().collect();
        for prefix in gone {
            self.withdraw(prefix).await?;
        }
        for route in &routes {
            self.announce(route).await?;
        }
        Ok(())
    }

    async fn announce(&mut self, route: &LocalRoute) -> Result<(), Finished> {
        let session = &self.session;
        if !session.families.carries(&route.prefix) {
            debug!(
                peer = %session.peer,
                prefix = %route.prefix,
                "Skipping route: its address family was not negotiated"
            );
            return Ok(());
        }
        let limits = session.export_limits.as_deref();
        if limits.is_some_and(|limits| !limits.allows(&route.prefix)) {
            debug!(
                peer = %session.peer,
                prefix = %route.prefix,
                "Not announcing route: its prefix is not in the peer's export limits"
            );
            return Ok(());
        }
        // What the peer would hold with this route.
        let routes = self.own_announced.len()
            + usize::from(!self.own_announced.contains(&route.prefix))
            + self.learned.as_ref().map_or(0, Learned::held_by_peer);
        if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) {
            return Err(over_limit(session, limit, routes));
        }
        let towards = self.towards();
        let Some(mut update) = build_announce(route, &towards) else {
            debug!(
                peer = %session.peer,
                prefix = %route.prefix,
                "Skipping route: address family does not match session transport"
            );
            return Ok(());
        };
        // What every route on its way to this peer goes through: whether
        // its communities let it go there, and what it is marked with.
        let egress = Self::egress(session, self.control, self.cluster_id, towards);
        if !egress.dress(&mut update.path_attributes) {
            debug!(
                peer = %session.peer,
                prefix = %route.prefix,
                "Not announcing route: its communities say not to this peer"
            );
            return Ok(());
        }
        export::finish(&mut update.path_attributes, &towards);
        debug!(peer = %session.peer, prefix = %route.prefix, "Announcing local route");
        session.send(Message::Update(update)).await?;
        self.own_announced.insert(route.prefix);
        self.note_advertised();
        Ok(())
    }

    async fn withdraw(&mut self, prefix: IpPrefix) -> Result<(), Finished> {
        if !self.own_announced.remove(&prefix) {
            // The peer was never told of it.
            return Ok(());
        }
        debug!(peer = %self.session.peer, %prefix, "Withdrawing local route");
        self.session
            .send(Message::Update(build_withdraw(
                &prefix,
                self.session.extended_next_hop,
            )))
            .await?;
        self.note_advertised();
        Ok(())
    }
}

/// Ends `session` because `routes` routes were to be announced on it, over
/// its export limit of `limit`; none of them has been.
fn over_limit(session: &SessionHandle, limit: u32, routes: usize) -> Finished {
    error!(
        peer = %session.peer,
        name = session.name.as_deref().unwrap_or_default(),
        routes,
        limit,
        "More routes were to be advertised to the peer than its export limit allows; advertising \
         none of them, ending the session and keeping the peer down"
    );
    session.end(SessionError::ExportLimitExceeded { limit, routes });
    Finished
}

/// What about a session decides how a route is announced on it.
#[derive(Debug, Clone, Copy)]
pub struct Towards {
    pub local_asn: u32,
    /// Our address on the session, which is the routes' next hop.
    pub next_hop: IpAddr,
    /// Whether the session negotiated four-octet AS numbers. If it did
    /// not, and an AS number in the path needs four octets, the path is
    /// also sent as AS4_PATH so that it survives the two-octet speaker
    /// (RFC 6793 §4.2.2).
    pub four_octet_asn: bool,
    /// 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 session takes IPv4 routes with IPv6 next hops
    /// (RFC 8950).
    pub extended_next_hop: bool,
}

/// Builds the UPDATE announcing `route` on a session.
///
/// Returns `None` when the route cannot be given the session's address as
/// next hop: an IPv6 route on a session over IPv4, or an IPv4 route on a
/// session over IPv6 that did not agree on RFC 8950.
pub fn build_announce(route: &LocalRoute, towards: &Towards) -> Option<UpdateMessage> {
    let attrs = &route.attrs;

    // Our AS number goes on the path as a route leaves the AS, and not
    // before (RFC 4271 §5.1.2): a route we originate has an empty path
    // inside it. A peer in our AS that found our AS number on the path
    // would take the route for one that had looped.
    let as_path = AsPathAttribute {
        segments: if towards.ibgp {
            vec![]
        } else {
            vec![AsPathSegment {
                ordered: true,
                path: vec![towards.local_asn; 1 + usize::from(attrs.prepend)],
            }]
        },
    };
    let as4_path = if towards.four_octet_asn {
        None
    } else {
        as_path.as4_path()
    };

    // Path attributes in ascending type-code order, as most implementations
    // expect them.
    let mut path_attributes = vec![
        PathAttribute::Origin(attrs.origin),
        PathAttribute::AsPath(as_path),
    ];
    // IPv4 routes go in the body of the UPDATE with their next hop in
    // NEXT_HOP; IPv6 routes go in MP_REACH_NLRI, which carries its own.
    let mut announced_routes = vec![];
    let mp_reach = match (route.prefix.address_family(), towards.next_hop) {
        (IpVersion::Ipv4, IpAddr::V4(next_hop)) => {
            path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(next_hop)));
            announced_routes.push(route.prefix.into());
            None
        }
        (IpVersion::Ipv6, IpAddr::V6(next_hop)) => {
            Some(PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
                family: AfiSafi::IPV6_UNICAST,
                next_hop: ipv6_next_hop(next_hop),
                prefixes: vec![route.prefix.into()],
            }))
        }
        // An IPv4 route over a link that has only IPv6 addresses: in
        // MP_REACH_NLRI like an IPv6 route, with the IPv6 next hop
        // (RFC 8950 §4).
        (IpVersion::Ipv4, IpAddr::V6(next_hop)) if towards.extended_next_hop => {
            Some(PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
                family: AfiSafi::IPV4_UNICAST,
                next_hop: ipv6_next_hop(next_hop),
                prefixes: vec![route.prefix.into()],
            }))
        }
        _ => return None,
    };
    if let Some(med) = attrs.med {
        path_attributes.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(
            med,
        )));
    }
    // LOCAL_PREF is sent to peers in our AS and to no others (RFC 4271
    // §5.1.5).
    if towards.ibgp {
        path_attributes.push(PathAttribute::LocalPref(LocalPrefPathAttribute(
            DEFAULT_LOCAL_PREF,
        )));
    }
    if !attrs.communities.is_empty() {
        path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute(
            attrs.communities.clone(),
        )));
    }
    path_attributes.extend(mp_reach);
    path_attributes.extend(as4_path);
    if !attrs.large_communities.is_empty() {
        path_attributes.push(PathAttribute::LargeCommunities(
            LargeCommunitiesPathAttribute {
                communities: attrs.large_communities.clone(),
            },
        ));
    }
    // A route that goes down to a customer or across to a peer may not
    // come up again from there: it is marked, for every network after us
    // to see (RFC 9234 §5, egress, 1). Routes we originate carry no such
    // mark of anyone else's. When routes learned from peers are passed
    // on, one that carries the mark must not go to a provider, a peer or
    // a route server at all (egress, 2;
    // `PeerRole::may_be_sent_only_to_customer_routes`).
    if let Some(marked_by) = towards
        .role
        .filter(|_| !towards.ibgp)
        .and_then(|role| role.only_to_customer_on_export(towards.local_asn))
    {
        path_attributes.push(PathAttribute::OnlyToCustomer(marked_by));
    }

    Some(UpdateMessage {
        withdrawn_routes: vec![],
        path_attributes,
        announced_routes,
    })
}

/// Our address `ours` as the next hop of a route.
///
/// A link-local address, which is all there is on a session between such
/// addresses, goes in both of the places MP_REACH_NLRI has: where the
/// global address would be and where the link-local one is (RFC 2545 §3
/// has no form for a link-local address alone). FRR sends it so, and takes
/// the route for unreachable if it comes by itself; BIRD takes either.
pub(crate) fn ipv6_next_hop(ours: Ipv6Addr) -> NlriNextHop {
    if ours.is_unicast_link_local() {
        NlriNextHop::Ipv6WithLl {
            global: ours,
            link_local: ours,
        }
    } else {
        NlriNextHop::Ipv6(ours)
    }
}

/// Builds the UPDATE withdrawing `prefix`. With `extended_next_hop` the
/// session has IPv4 routes announced in MP_REACH_NLRI (RFC 8950), and they
/// are withdrawn the way they were announced.
pub fn build_withdraw(prefix: &IpPrefix, extended_next_hop: bool) -> UpdateMessage {
    match prefix.address_family() {
        IpVersion::Ipv4 if extended_next_hop => UpdateMessage {
            withdrawn_routes: vec![],
            path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
                family: AfiSafi::IPV4_UNICAST,
                prefixes: vec![Nlri::from(*prefix)],
            })],
            announced_routes: vec![],
        },
        IpVersion::Ipv4 => UpdateMessage {
            withdrawn_routes: vec![Nlri::from(*prefix)],
            path_attributes: vec![],
            announced_routes: vec![],
        },
        IpVersion::Ipv6 => UpdateMessage {
            withdrawn_routes: vec![],
            path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
                family: AfiSafi::IPV6_UNICAST,
                prefixes: vec![Nlri::from(*prefix)],
            })],
            announced_routes: vec![],
        },
    }
}

#[cfg(test)]
mod tests {
    use std::net::{Ipv4Addr, Ipv6Addr};

    use bgp_packet::message::{LargeCommunity, OriginPathAttribute};
    use bgp_primitives::parser::ParserContext;
    use bgp_rib::LocalRouteAttrs;
    use bytes::BytesMut;

    use super::*;

    fn prefix(s: &str) -> IpPrefix {
        IpPrefix::try_from(s).unwrap()
    }

    /// On a session between link-local addresses ours is the next hop,
    /// written the way FRR writes its own and takes ours.
    #[test]
    fn a_link_local_next_hop_is_given_twice() {
        let ours: Ipv6Addr = "fe80::1".parse().unwrap();
        let towards = Towards {
            extended_next_hop: true,
            ..towards(65000, IpAddr::V6(ours), true)
        };
        for (route, family) in [
            ("10.1.0.0/16", AfiSafi::IPV4_UNICAST),
            ("2001:db8:1::/48", AfiSafi::IPV6_UNICAST),
        ] {
            let local = LocalRoute::new(prefix(route), LocalRouteAttrs::default());
            let update = build_announce(&local, &towards).unwrap();
            assert!(update.path_attributes.contains(&PathAttribute::MpReachNlri(
                MpReachNlriPathAttribute {
                    family,
                    next_hop: NlriNextHop::Ipv6WithLl {
                        global: ours,
                        link_local: ours,
                    },
                    prefixes: vec![prefix(route).into()],
                }
            )));
        }
    }

    /// RFC 8950: over a link that has only IPv6 addresses, an IPv4 route
    /// goes in MP_REACH_NLRI with our IPv6 address as next hop, and is
    /// withdrawn in MP_UNREACH_NLRI.
    #[test]
    fn an_ipv4_route_goes_over_an_ipv6_session_if_both_agreed() {
        let route = LocalRoute::new(prefix("10.1.0.0/16"), LocalRouteAttrs::default());
        let ours: Ipv6Addr = "2001:db8::1".parse().unwrap();
        let over_ipv6 = towards(65000, IpAddr::V6(ours), true);
        // Not agreed: the route has no next hop the peer could use.
        assert!(build_announce(&route, &over_ipv6).is_none());

        let agreed = Towards {
            extended_next_hop: true,
            ..over_ipv6
        };
        let update = build_announce(&route, &agreed).unwrap();
        assert!(update.announced_routes.is_empty(), "not in the body");
        assert!(
            !update
                .path_attributes
                .iter()
                .any(|attribute| matches!(attribute, PathAttribute::NextHop(_))),
            "no NEXT_HOP attribute"
        );
        assert!(update.path_attributes.contains(&PathAttribute::MpReachNlri(
            MpReachNlriPathAttribute {
                family: AfiSafi::IPV4_UNICAST,
                next_hop: NlriNextHop::Ipv6(ours),
                prefixes: vec![prefix("10.1.0.0/16").into()],
            }
        )));
        // What is written is what a peer reads.
        let ctx = ParserContext {
            four_octet_asn: true,
            ibgp: false,
            add_path: Vec::new(),
        };
        let mut wire = BytesMut::new();
        Message::Update(update.clone())
            .to_wire(&ctx, &mut wire)
            .unwrap();
        assert_eq!(
            Message::decode(&ctx, &wire).unwrap(),
            Message::Update(update)
        );

        let withdrawal = build_withdraw(&prefix("10.1.0.0/16"), true);
        assert!(withdrawal.withdrawn_routes.is_empty());
        assert_eq!(
            withdrawal.path_attributes,
            [PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
                family: AfiSafi::IPV4_UNICAST,
                prefixes: vec![prefix("10.1.0.0/16").into()],
            })]
        );
        // An IPv6 route is as it always was, agreed or not.
        let v6 = LocalRoute::new(prefix("2001:db8:1::/48"), LocalRouteAttrs::default());
        assert_eq!(
            build_announce(&v6, &agreed),
            build_announce(&v6, &over_ipv6)
        );
    }

    /// RFC 9234 §5, the egress procedure for routes we originate.
    #[test]
    fn a_route_is_marked_only_to_customer_on_its_way_down_or_across() {
        let route = LocalRoute::new(
            IpPrefix::try_from("10.1.0.0/16").unwrap(),
            LocalRouteAttrs::default(),
        );
        let to = |role| Towards {
            role,
            ..towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true)
        };
        let mark = |role| {
            build_announce(&route, &to(role))
                .unwrap()
                .path_attributes
                .into_iter()
                .find(|attribute| matches!(attribute, PathAttribute::OnlyToCustomer(_)))
        };
        for role in [
            PeerRole::Customer,
            PeerRole::Peer,
            PeerRole::RouteServerClient,
        ] {
            assert_eq!(
                mark(Some(role)),
                Some(PathAttribute::OnlyToCustomer(65000)),
                "{role:?}"
            );
        }
        // On its way up it may still go anywhere; and where it is not
        // known what the peer is, nothing is said.
        for role in [Some(PeerRole::Provider), Some(PeerRole::RouteServer), None] {
            assert_eq!(mark(role), None, "{role:?}");
        }
    }

    /// An external peer.
    fn towards(local_asn: u32, next_hop: IpAddr, four_octet_asn: bool) -> Towards {
        Towards {
            local_asn,
            next_hop,
            four_octet_asn,
            ibgp: false,
            role: None,
            extended_next_hop: false,
        }
    }

    #[test]
    fn ipv4_announce_has_expected_attributes_and_roundtrips() {
        let route = LocalRoute::new(
            prefix("10.100.1.0/24"),
            LocalRouteAttrs {
                origin: OriginPathAttribute::IGP,
                prepend: 2,
                med: Some(50),
                communities: vec![(65000, 100), (65535, 65281)],
                large_communities: vec![LargeCommunity {
                    global_admin: 65000,
                    data_1: 1,
                    data_2: 2,
                }],
            },
        );
        let update = build_announce(
            &route,
            &towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true),
        )
        .unwrap();

        assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]);
        assert!(update.withdrawn_routes.is_empty());
        assert_eq!(
            update.path_attributes,
            vec![
                PathAttribute::Origin(OriginPathAttribute::IGP),
                AsPathAttribute::from_asns(vec![65000, 65000, 65000]),
                PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
                PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(50)),
                PathAttribute::Communities(CommunitiesPathAttribute(vec![
                    (65000, 100),
                    (65535, 65281)
                ])),
                PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute {
                    communities: vec![LargeCommunity {
                        global_admin: 65000,
                        data_1: 1,
                        data_2: 2
                    }],
                }),
            ]
        );

        // Serialise and parse back to make sure the wire encoding is sound.
        let ctx = ParserContext {
            four_octet_asn: true,
            ibgp: false,
            add_path: Vec::new(),
        };
        let msg = Message::Update(update.clone());
        let mut buf = BytesMut::new();
        msg.to_wire(&ctx, &mut buf).unwrap();
        let Message::Update(parsed) = Message::decode(&ctx, &buf).unwrap() else {
            panic!("expected UPDATE")
        };
        assert_eq!(parsed, update);
    }

    #[test]
    fn ipv6_announce_uses_mp_reach() {
        let route = LocalRoute::new(prefix("2001:db8:1::/48"), LocalRouteAttrs::default());
        let nh = Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1);
        let update = build_announce(&route, &towards(65000, IpAddr::V6(nh), true)).unwrap();

        assert!(update.announced_routes.is_empty());
        assert!(
            !update
                .path_attributes
                .iter()
                .any(|a| matches!(a, PathAttribute::NextHop(_)))
        );
        assert!(update.path_attributes.iter().any(|a| matches!(
            a,
            PathAttribute::MpReachNlri(MpReachNlriPathAttribute { next_hop: NlriNextHop::Ipv6(h), prefixes, .. })
                if *h == nh && *prefixes == vec![prefix("2001:db8:1::/48")]
        )));
    }

    /// A two-octet session cannot carry a four-octet AS number in AS_PATH:
    /// it goes out as AS_TRANS there, and in full in AS4_PATH.
    #[test]
    fn a_wide_as_number_reaches_a_two_octet_peer_in_as4_path() {
        let route = LocalRoute::new(
            prefix("10.100.1.0/24"),
            LocalRouteAttrs {
                prepend: 1,
                ..LocalRouteAttrs::default()
            },
        );
        let next_hop = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1));
        let path = |update: &UpdateMessage, as4: bool| {
            update.path_attributes.iter().find_map(|a| match a {
                PathAttribute::As4Path(p) if as4 => Some(p.segments[0].path.clone()),
                PathAttribute::AsPath(p) if !as4 => Some(p.segments[0].path.clone()),
                _ => None,
            })
        };

        let update = build_announce(&route, &towards(4_200_000_001, next_hop, false)).unwrap();
        assert_eq!(path(&update, true), Some(vec![4_200_000_001; 2]));

        // What the peer reads from the wire.
        let ctx = ParserContext {
            four_octet_asn: false,
            ibgp: false,
            add_path: Vec::new(),
        };
        let mut buf = BytesMut::new();
        Message::Update(update).to_wire(&ctx, &mut buf).unwrap();
        let Message::Update(seen) = Message::decode(&ctx, &buf).unwrap() else {
            panic!("expected UPDATE")
        };
        assert_eq!(path(&seen, false), Some(vec![23456; 2]), "AS_TRANS");
        assert_eq!(path(&seen, true), Some(vec![4_200_000_001; 2]));

        // No AS4_PATH when it would say nothing new: the peer handles
        // four-octet numbers, or ours fits in two.
        let update = build_announce(&route, &towards(4_200_000_001, next_hop, true)).unwrap();
        assert_eq!(path(&update, true), None);
        let update = build_announce(&route, &towards(65000, next_hop, false)).unwrap();
        assert_eq!(path(&update, true), None);
        assert_eq!(path(&update, false), Some(vec![65000; 2]));
    }

    #[test]
    fn family_mismatch_is_skipped() {
        let v6 = LocalRoute::new(prefix("2001:db8::/32"), LocalRouteAttrs::default());
        assert!(
            build_announce(&v6, &towards(65000, IpAddr::V4(Ipv4Addr::LOCALHOST), true)).is_none()
        );
        let v4 = LocalRoute::new(prefix("10.0.0.0/8"), LocalRouteAttrs::default());
        assert!(
            build_announce(&v4, &towards(65000, IpAddr::V6(Ipv6Addr::LOCALHOST), true)).is_none()
        );
    }

    #[test]
    fn withdraw_by_family() {
        let v4 = build_withdraw(&prefix("10.0.0.0/8"), false);
        assert_eq!(v4.withdrawn_routes, vec![prefix("10.0.0.0/8")]);
        assert!(v4.path_attributes.is_empty());

        let v6 = build_withdraw(&prefix("2001:db8::/32"), false);
        assert!(v6.withdrawn_routes.is_empty());
        assert!(
            matches!(&v6.path_attributes[..], [PathAttribute::MpUnreachNlri(m)] if m.prefixes == vec![prefix("2001:db8::/32")])
        );
    }

    /// Inside our own AS a route we originate has an empty AS path and a
    /// LOCAL_PREF. With our AS number on the path, a peer in our AS takes
    /// the route for a loop; without LOCAL_PREF it treats the route as
    /// withdrawn, as our own decoder does.
    #[test]
    fn a_route_announced_inside_our_as_has_an_empty_path_and_a_local_pref() {
        let route = LocalRoute::new(
            prefix("10.100.1.0/24"),
            LocalRouteAttrs {
                prepend: 3,
                ..LocalRouteAttrs::default()
            },
        );
        let internal = Towards {
            ibgp: true,
            ..towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true)
        };
        let update = build_announce(&route, &internal).unwrap();
        assert_eq!(
            update.path_attributes,
            vec![
                PathAttribute::Origin(OriginPathAttribute::IGP),
                PathAttribute::AsPath(AsPathAttribute { segments: vec![] }),
                PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
                PathAttribute::LocalPref(LocalPrefPathAttribute(100)),
            ]
        );

        // An internal peer accepts it as it stands.
        let ctx = ParserContext {
            four_octet_asn: true,
            ibgp: true,
            add_path: Vec::new(),
        };
        let mut buf = BytesMut::new();
        Message::Update(update.clone())
            .to_wire(&ctx, &mut buf)
            .unwrap();
        let decoded = Message::decode_with_report(&ctx, &buf).unwrap();
        assert!(
            decoded.update_errors.is_empty(),
            "{:?}",
            decoded.update_errors
        );
        assert_eq!(decoded.message, Message::Update(update));

        // To an external peer the same route has our AS on the path, as
        // many times as configured, and no LOCAL_PREF.
        let external = Towards {
            ibgp: false,
            ..internal
        };
        let update = build_announce(&route, &external).unwrap();
        assert_eq!(
            update.path_attributes[1],
            AsPathAttribute::from_asns(vec![65000; 4])
        );
        assert!(
            !update
                .path_attributes
                .iter()
                .any(|a| matches!(a, PathAttribute::LocalPref(_)))
        );
    }

    // -- Exporters --

    use std::time::Duration;

    use bgp_rib::table::{PeerInfo, RibTable};
    use tokio::sync::mpsc;

    use crate::collision::Direction;
    use crate::family::FamilySet;
    use crate::fsm::Negotiated;
    use crate::session::SessionGuard;

    /// A registry and local RIB with an advertiser running, and a table to
    /// take peer numbers from.
    struct Fixture {
        registry: SessionRegistry,
        local_rib: LocalRib,
        table: RibTable,
    }

    /// The far end of a session's queue: what the session's task would
    /// write to the peer.
    struct Queue {
        outbound: mpsc::Receiver<Message>,
        /// What the peer holds, going by what it has been sent.
        routes: HashSet<IpPrefix>,
        _registration: SessionGuard,
    }

    impl Fixture {
        fn new() -> Self {
            let registry = SessionRegistry::new();
            let local_rib = LocalRib::spawn();
            spawn_advertiser(
                local_rib.clone(),
                Arc::new(RibTable::new(4)),
                registry.clone(),
                65000,
                None,
                Ipv4Addr::new(192, 0, 2, 1),
                watch::channel(true).1,
            );
            Self {
                registry,
                local_rib,
                table: RibTable::new(1),
            }
        }

        /// A session with peer number `n`, whose queue holds `capacity`
        /// messages.
        fn session(&self, n: u8, capacity: usize) -> Queue {
            let address = IpAddr::V4(Ipv4Addr::new(192, 0, 2, n));
            let negotiated = Negotiated {
                peer_asn: 65000 + u32::from(n),
                ibgp: false,
                peer_identifier: Ipv4Addr::new(10, 0, 0, n),
                hold_time_secs: 90,
                four_octet_asn: true,
                route_refresh: false,
                families: FamilySet::supported(),
                add_path_receive: FamilySet::default(),
                add_path_send: FamilySet::default(),
                role_confirmed: false,
                extended_next_hop: false,
                graceful_restart: None,
            };
            let rib_peer = self
                .table
                .add_peer(PeerInfo {
                    asn: negotiated.peer_asn,
                    router_id: negotiated.peer_identifier,
                    address,
                    interface: 0,
                    ibgp: false,
                    reflector_client: false,
                })
                .unwrap();
            let (tx, outbound) = mpsc::channel(capacity);
            let handle = SessionHandle::new(
                PeerId {
                    addr: (address, 179).into(),
                    asn: negotiated.peer_asn,
                },
                None,
                Direction::Inbound,
                (Ipv4Addr::new(192, 0, 2, 100), 179).into(),
                &negotiated,
                rib_peer,
                tx,
            );
            Queue {
                outbound,
                routes: HashSet::new(),
                _registration: self.registry.register(handle),
            }
        }

        async fn announce(&self, n: u32) {
            let route = LocalRoute::new(nth_prefix(n), LocalRouteAttrs::default());
            self.local_rib.announce(route).await.unwrap();
        }

        async fn routes(&self) -> HashSet<IpPrefix> {
            let routes = self.local_rib.snapshot().await.unwrap();
            routes.iter().map(|route| route.prefix).collect()
        }
    }

    fn nth_prefix(n: u32) -> IpPrefix {
        let bytes = (0x0a00_0000 + (n << 8)).to_be_bytes();
        IpPrefix::new(IpVersion::Ipv4, &bytes[..3], 24).unwrap()
    }

    impl Queue {
        /// Reads what has been queued for the peer until it holds exactly
        /// `expected`, as a peer would that takes every message in turn.
        async fn read_until_it_has(&mut self, expected: &HashSet<IpPrefix>) {
            let reading = async {
                while self.routes != *expected {
                    let Some(Message::Update(update)) = self.outbound.recv().await else {
                        panic!("expected an UPDATE");
                    };
                    for nlri in &update.withdrawn_routes {
                        self.routes.remove(&nlri.prefix);
                    }
                    for nlri in &update.announced_routes {
                        self.routes.insert(nlri.prefix);
                    }
                }
            };
            if tokio::time::timeout(Duration::from_secs(10), reading)
                .await
                .is_err()
            {
                panic!(
                    "the peer has {} routes, {} of them right, and should have {}",
                    self.routes.len(),
                    self.routes.intersection(expected).count(),
                    expected.len()
                );
            }
        }
    }

    /// A route from a router of our AS reaches a route reflector client
    /// of ours through the advertiser, marked with where it started and our
    /// cluster, with its preference, and with us as next hop.
    #[tokio::test]
    async fn a_route_from_one_of_our_routers_is_reflected_to_a_client() {
        let registry = SessionRegistry::new();
        let rib = Arc::new(RibTable::new(4));
        let cluster = Ipv4Addr::new(10, 0, 0, 100);
        spawn_advertiser(
            LocalRib::spawn(),
            Arc::clone(&rib),
            registry.clone(),
            65000,
            None,
            cluster,
            watch::channel(true).1,
        );
        let ours = |n: u8, client| PeerInfo {
            asn: 65000,
            router_id: Ipv4Addr::new(10, 0, 0, n),
            address: IpAddr::V4(Ipv4Addr::new(192, 0, 2, n)),
            interface: 0,
            ibgp: true,
            reflector_client: client,
        };
        // Another router of ours, not a client, with a route in the table.
        let other = rib.add_peer(ours(1, false)).unwrap();
        let came = [
            PathAttribute::Origin(OriginPathAttribute::IGP),
            AsPathAttribute::from_asns(vec![64496]),
            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
            PathAttribute::LocalPref(LocalPrefPathAttribute(200)),
        ];
        let attrs = rib.attrs().intern(&came, None).unwrap();
        rib.announce(other, &attrs, &[Nlri::from(nth_prefix(1))]);

        // A client that is told of learned routes.
        let negotiated = Negotiated {
            peer_asn: 65000,
            ibgp: true,
            peer_identifier: Ipv4Addr::new(10, 0, 0, 2),
            hold_time_secs: 90,
            four_octet_asn: true,
            route_refresh: false,
            families: FamilySet::supported(),
            add_path_receive: FamilySet::default(),
            add_path_send: FamilySet::default(),
            role_confirmed: false,
            extended_next_hop: false,
            graceful_restart: None,
        };
        let rib_peer = rib.add_peer(ours(2, true)).unwrap();
        let (tx, mut outbound) = mpsc::channel(16);
        let mut handle = SessionHandle::new(
            PeerId {
                addr: (Ipv4Addr::new(192, 0, 2, 2), 179).into(),
                asn: 65000,
            },
            None,
            Direction::Inbound,
            (Ipv4Addr::new(192, 0, 2, 100), 179).into(),
            &negotiated,
            rib_peer,
            tx,
        );
        handle.export = Arc::new(crate::policy::ExportPolicy {
            learned_routes: true,
            all_paths: false,
            ..crate::policy::ExportPolicy::default()
        });
        handle.reflector_client = true;
        let _registration = registry.register(handle);

        let update = tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                match outbound.recv().await {
                    Some(Message::Update(update)) if !update.announced_routes.is_empty() => {
                        return update;
                    }
                    Some(_) => {}
                    None => panic!("the session's queue closed"),
                }
            }
        })
        .await
        .expect("the client is told of the route");
        assert_eq!(update.announced_routes, [Nlri::from(nth_prefix(1))]);
        assert_eq!(
            update.path_attributes,
            [
                PathAttribute::Origin(OriginPathAttribute::IGP),
                AsPathAttribute::from_asns(vec![64496]),
                PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 100))),
                PathAttribute::LocalPref(LocalPrefPathAttribute(200)),
                PathAttribute::OriginatorId(Ipv4Addr::new(10, 0, 0, 1)),
                PathAttribute::ClusterList(vec![cluster]),
            ]
        );
    }

    /// One peer that takes nothing must not keep the others from being
    /// sent their routes. There used to be one task announcing to every
    /// session in turn, and it waited for whichever queue was full.
    #[tokio::test]
    async fn a_peer_that_reads_nothing_holds_up_nobody_else() {
        let fixture = Fixture::new();
        let _stuck = fixture.session(1, 4);
        let mut reading = fixture.session(2, 4);

        for n in 0..500 {
            fixture.announce(n).await;
        }
        reading.read_until_it_has(&fixture.routes().await).await;
        assert_eq!(reading.routes.len(), 500);
    }

    /// A peer that reads so slowly that changes pass it by is brought up
    /// to date when it reads again: with the routes there now are, and
    /// without the ones that went while it was not looking. Those used to
    /// stay announced to it for good.
    #[tokio::test]
    async fn a_peer_that_fell_behind_ends_up_with_the_routes_there_are() {
        let fixture = Fixture::new();
        let mut slow = fixture.session(1, 4);
        for n in 0..10 {
            fixture.announce(n).await;
        }
        slow.read_until_it_has(&fixture.routes().await).await;

        // Far more changes than are kept for a subscriber that is not
        // reading, among them the withdrawal of what the peer has.
        for n in 0..10 {
            fixture.local_rib.withdraw(nth_prefix(n)).await.unwrap();
        }
        for n in 10..3000 {
            fixture.announce(n).await;
        }
        for n in 2000..3000 {
            fixture.local_rib.withdraw(nth_prefix(n)).await.unwrap();
        }

        let routes = fixture.routes().await;
        assert_eq!(routes.len(), 1990);
        slow.read_until_it_has(&routes).await;
    }

    /// When a peer's session is replaced by a new one, the new session is
    /// sent the routes from the beginning.
    #[tokio::test]
    async fn a_new_session_with_the_same_peer_starts_afresh() {
        let fixture = Fixture::new();
        fixture.announce(1).await;
        let routes = fixture.routes().await;

        let mut first = fixture.session(1, 4);
        first.read_until_it_has(&routes).await;
        drop(first);
        let mut second = fixture.session(1, 4);
        second.read_until_it_has(&routes).await;
    }

    // -- Export limits --

    use crate::policy::{ExportLimits, ExportPolicy, PrefixSet};

    /// A table with routes from a customer for `prefixes`, an advertiser
    /// over it with `originated` routes of our own, and a session to a
    /// peer of ours that is told of learned routes, within `limits`.
    async fn limited(
        prefixes: &[IpPrefix],
        originated: &[IpPrefix],
        limits: ExportLimits,
    ) -> LimitedPeer {
        let registry = SessionRegistry::new();
        let rib = Arc::new(RibTable::new(4));
        let local_rib = LocalRib::spawn();
        for prefix in originated {
            let route = LocalRoute::new(*prefix, LocalRouteAttrs::default());
            local_rib.announce(route).await.unwrap();
        }
        spawn_advertiser(
            local_rib,
            Arc::clone(&rib),
            registry.clone(),
            65000,
            None,
            Ipv4Addr::new(192, 0, 2, 1),
            watch::channel(true).1,
        );
        let peer = |n: u8, asn| PeerInfo {
            asn,
            router_id: Ipv4Addr::new(10, 0, 0, n),
            address: IpAddr::V4(Ipv4Addr::new(192, 0, 2, n)),
            interface: 0,
            ibgp: false,
            reflector_client: false,
        };
        // A customer's routes carry no Only to Customer mark: as far as
        // the roles go, they are for everyone.
        let customer = rib.add_peer(peer(1, 64496)).unwrap();
        let came = [
            PathAttribute::Origin(OriginPathAttribute::IGP),
            AsPathAttribute::from_asns(vec![64496]),
            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
        ];
        let attrs = rib.attrs().intern(&came, None).unwrap();
        let nlri: Vec<Nlri> = prefixes.iter().copied().map(Nlri::from).collect();
        rib.announce(customer, &attrs, &nlri);

        let negotiated = Negotiated {
            peer_asn: 64497,
            ibgp: false,
            peer_identifier: Ipv4Addr::new(10, 0, 0, 2),
            hold_time_secs: 90,
            four_octet_asn: true,
            route_refresh: false,
            families: FamilySet::supported(),
            add_path_receive: FamilySet::default(),
            add_path_send: FamilySet::default(),
            role_confirmed: false,
            extended_next_hop: false,
            graceful_restart: None,
        };
        let rib_peer = rib.add_peer(peer(2, 64497)).unwrap();
        let (tx, outbound) = mpsc::channel(1024);
        let mut handle = SessionHandle::new(
            PeerId {
                addr: (Ipv4Addr::new(192, 0, 2, 2), 179).into(),
                asn: 64497,
            },
            None,
            Direction::Inbound,
            (Ipv4Addr::new(192, 0, 2, 100), 179).into(),
            &negotiated,
            rib_peer,
            tx,
        );
        handle.role = Some(PeerRole::Peer);
        handle.export = Arc::new(ExportPolicy {
            learned_routes: true,
            all_paths: false,
            ..ExportPolicy::default()
        });
        handle.export_limits = Some(Arc::new(limits));
        let ending = handle.listen_for_end();
        LimitedPeer {
            outbound,
            ending,
            _registration: registry.register(handle),
        }
    }

    /// The far end of the limited peer's session: what it is sent, and
    /// what ends it.
    struct LimitedPeer {
        outbound: mpsc::Receiver<Message>,
        ending: mpsc::UnboundedReceiver<SessionError>,
        _registration: SessionGuard,
    }

    /// The routes announced on a session's queue until it has been quiet
    /// for a while.
    async fn announced(outbound: &mut mpsc::Receiver<Message>) -> HashSet<IpPrefix> {
        let mut routes = HashSet::new();
        while let Ok(Some(message)) =
            tokio::time::timeout(Duration::from_millis(500), outbound.recv()).await
        {
            if let Message::Update(update) = message {
                for nlri in &update.withdrawn_routes {
                    routes.remove(&nlri.prefix);
                }
                routes.extend(update.announced_routes.iter().map(|nlri| nlri.prefix));
                if let Some(PathAttribute::MpReachNlri(reach)) = update
                    .path_attributes
                    .iter()
                    .find(|a| matches!(a, PathAttribute::MpReachNlri(_)))
                {
                    routes.extend(reach.prefixes.iter().map(|nlri| nlri.prefix));
                }
            }
        }
        routes
    }

    fn allowing(prefixes: &[IpPrefix]) -> Option<PrefixSet> {
        let mut set = PrefixSet::default();
        for prefix in prefixes {
            set.allow(*prefix, prefix.length());
        }
        Some(set)
    }

    /// Only routes for the prefixes in the export limits go to a peer,
    /// learned or our own, though nothing else keeps the others from it:
    /// a customer's routes are for everyone, as far as the roles go.
    #[tokio::test]
    async fn routes_outside_the_export_limits_do_not_go_to_a_peer() {
        let limits = ExportLimits {
            prefixes: allowing(&[nth_prefix(2), nth_prefix(11)]),
            max_prefixes: None,
        };
        let mut peer = limited(
            &[nth_prefix(1), nth_prefix(2), nth_prefix(3)],
            &[nth_prefix(10), nth_prefix(11)],
            limits,
        )
        .await;
        assert_eq!(
            announced(&mut peer.outbound).await,
            HashSet::from([nth_prefix(2), nth_prefix(11)])
        );
        assert!(peer.ending.try_recv().is_err(), "the session goes on");
    }

    /// When more routes would go to a peer than its export limit allows,
    /// none of them does, and the session is ended: a leak becomes a
    /// session that is down, and not a table spread across the Internet.
    #[tokio::test]
    async fn more_routes_than_the_export_limit_end_the_session_and_none_goes() {
        let prefixes: Vec<IpPrefix> = (1..=5).map(nth_prefix).collect();
        let limits = ExportLimits {
            prefixes: None,
            max_prefixes: Some(3),
        };
        let mut peer = limited(&prefixes, &[], limits).await;
        let why = tokio::time::timeout(Duration::from_secs(10), peer.ending.recv())
            .await
            .expect("the session is asked to end")
            .expect("by the exporter");
        assert!(
            matches!(
                why,
                SessionError::ExportLimitExceeded {
                    limit: 3,
                    routes: 5
                }
            ),
            "{why:?}"
        );
        assert_eq!(announced(&mut peer.outbound).await, HashSet::new());
    }

    /// Our own routes count too, and none goes past the limit: those that
    /// fit go, and the one that does not ends the session.
    #[tokio::test]
    async fn originated_routes_count_towards_the_export_limit() {
        let originated: Vec<IpPrefix> = (10..13).map(nth_prefix).collect();
        let limits = ExportLimits {
            prefixes: None,
            max_prefixes: Some(2),
        };
        let mut peer = limited(&[], &originated, limits).await;
        let why = tokio::time::timeout(Duration::from_secs(10), peer.ending.recv())
            .await
            .expect("the session is asked to end")
            .expect("by the exporter");
        assert!(
            matches!(
                why,
                SessionError::ExportLimitExceeded {
                    limit: 2,
                    routes: 3
                }
            ),
            "{why:?}"
        );
        assert_eq!(announced(&mut peer.outbound).await.len(), 2);
    }
}