From 666471e38df2e68f7aa320f906f9b1a846a28d90 Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Thu, 1 Oct 2026 21:08:14 +0200 Subject: [PATCH] Share path attributes between Adj-RIB-In entries Each received prefix kept its own copy of the UPDATE's whole attribute list, and for multiprotocol routes that list included MP_REACH_NLRI with every other prefix from the same UPDATE: n prefixes stored n copies of n prefixes. An UPDATE is now turned into a ReceivedUpdate once. Its prefixes share one attribute list, which leaves out MP_REACH_NLRI and MP_UNREACH_NLRI; the next hop from MP_REACH_NLRI is kept on the entry. The RIB takes one command per UPDATE instead of one per prefix, and its table is keyed by peer so that dropping a peer's routes no longer scans every route. The status API shows the multiprotocol next hop as an address instead of its Debug form. Co-Authored-By: Claude Fable 5.1 --- crates/packet/src/message.rs | 12 +++ crates/rib/src/adj_rib_in.rs | 177 +++++++++++++++++++++++------------ crates/rib/src/entry.rs | 177 ++++++++++++++++++++++++++++++++--- crates/rib/src/lib.rs | 2 +- crates/server/src/fsm.rs | 52 +--------- crates/web/src/api.rs | 21 +++-- crates/web/src/events.rs | 4 +- 7 files changed, 315 insertions(+), 130 deletions(-) diff --git a/crates/packet/src/message.rs b/crates/packet/src/message.rs index de8863c..57b2890 100644 --- a/crates/packet/src/message.rs +++ b/crates/packet/src/message.rs @@ -1227,6 +1227,18 @@ pub enum NlriNextHop { }, } +impl std::fmt::Display for NlriNextHop { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + NlriNextHop::Ipv4(addr) => write!(f, "{addr}"), + NlriNextHop::Ipv6(addr) => write!(f, "{addr}"), + NlriNextHop::Ipv6WithLl { global, link_local } => { + write!(f, "{global} (link-local {link_local})") + } + } + } +} + impl NlriNextHop { pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<()> { match self { diff --git a/crates/rib/src/adj_rib_in.rs b/crates/rib/src/adj_rib_in.rs index 20bb0bb..d718a5d 100644 --- a/crates/rib/src/adj_rib_in.rs +++ b/crates/rib/src/adj_rib_in.rs @@ -1,24 +1,25 @@ +//! The Adj-RIB-In: routes as peers announced them, before any policy. +//! +//! One actor owns the table. Handles are cheap to clone and talk to it over +//! a channel; changes are published as [`RibEvent`]s. + use std::collections::HashMap; use bgp_primitives::ip_prefix::IpPrefix; use tokio::sync::{broadcast, mpsc, oneshot}; use tracing::warn; -use crate::entry::{PeerId, RibEntry, SnapshotEntry}; +use crate::entry::{PeerId, ReceivedUpdate, RibEntry, SnapshotEntry}; use crate::event::RibEvent; const EVENT_CHANNEL_CAPACITY: usize = 1024; const CMD_CHANNEL_CAPACITY: usize = 256; enum RibCommand { - Announce { + /// Apply one UPDATE from `peer`. + Apply { peer: PeerId, - prefix: IpPrefix, - entry: RibEntry, - }, - Withdraw { - peer: PeerId, - prefix: IpPrefix, + update: ReceivedUpdate, }, /// Drop everything learned from `peer`, e.g. because its session ended. RemovePeer { @@ -30,8 +31,7 @@ enum RibCommand { }, } -/// Handle to the Adj-RIB-In actor. Cheap to clone — all clones share the same -/// backing task. The actor is stopped when all handles are dropped. +/// A cheaply-cloneable handle to the Adj-RIB-In actor. #[derive(Clone)] pub struct AdjRibIn { tx: mpsc::Sender, @@ -43,24 +43,15 @@ impl AdjRibIn { pub fn spawn() -> Self { let (tx, rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); - let events_clone = events.clone(); - tokio::spawn(run_actor(rx, events_clone)); + tokio::spawn(run_actor(rx, events.clone())); Self { tx, events } } - pub async fn announce(&self, peer: PeerId, prefix: IpPrefix, entry: RibEntry) { - let _ = self - .tx - .send(RibCommand::Announce { - peer, - prefix, - entry, - }) - .await; - } - - pub async fn withdraw(&self, peer: PeerId, prefix: IpPrefix) { - let _ = self.tx.send(RibCommand::Withdraw { peer, prefix }).await; + /// Applies one UPDATE received from `peer`: its withdrawals first, then + /// its announcements, so a prefix named in both ends up announced + /// (RFC 4271 §4.3). + pub async fn apply(&self, peer: PeerId, update: ReceivedUpdate) { + let _ = self.tx.send(RibCommand::Apply { peer, update }).await; } /// Removes every route learned from `peer`, emitting a @@ -96,50 +87,49 @@ impl AdjRibIn { } async fn run_actor(mut rx: mpsc::Receiver, events: broadcast::Sender) { - let mut table: HashMap<(PeerId, IpPrefix), RibEntry> = HashMap::new(); + // Keyed by peer first, so that dropping a peer's routes when its session + // ends touches only that peer's table. + let mut table: HashMap> = HashMap::new(); while let Some(cmd) = rx.recv().await { match cmd { - // Send errors below mean there are no subscribers, which is fine. - RibCommand::Announce { - peer, - prefix, - entry, - } => { - table.insert((peer, prefix.clone()), entry.clone()); - let _ = events.send(RibEvent::Announced { - peer, - prefix, - entry, - }); - } - RibCommand::Withdraw { peer, prefix } => { - if table.remove(&(peer, prefix.clone())).is_some() { - let _ = events.send(RibEvent::Withdrawn { peer, prefix }); + // Event send errors below mean there are no subscribers, which is fine. + RibCommand::Apply { peer, update } => { + let routes = table.entry(peer).or_default(); + for prefix in update.withdrawn { + if routes.remove(&prefix).is_some() { + let _ = events.send(RibEvent::Withdrawn { peer, prefix }); + } + } + for (prefix, entry) in update.announced { + routes.insert(prefix.clone(), entry.clone()); + let _ = events.send(RibEvent::Announced { + peer, + prefix, + entry, + }); + } + if routes.is_empty() { + table.remove(&peer); } } RibCommand::RemovePeer { peer, reply } => { - let prefixes: Vec = table - .keys() - .filter(|(p, _)| *p == peer) - .map(|(_, prefix)| prefix.clone()) - .collect(); - for prefix in &prefixes { - table.remove(&(peer, prefix.clone())); - let _ = events.send(RibEvent::Withdrawn { - peer, - prefix: prefix.clone(), - }); + let routes = table.remove(&peer).unwrap_or_default(); + let removed = routes.len(); + for prefix in routes.into_keys() { + let _ = events.send(RibEvent::Withdrawn { peer, prefix }); } - let _ = reply.send(prefixes.len()); + let _ = reply.send(removed); } RibCommand::Snapshot { reply } => { let snap = table .iter() - .map(|((peer, prefix), entry)| SnapshotEntry { - peer: *peer, - prefix: prefix.clone(), - entry: entry.clone(), + .flat_map(|(peer, routes)| { + routes.iter().map(|(prefix, entry)| SnapshotEntry { + peer: *peer, + prefix: prefix.clone(), + entry: entry.clone(), + }) }) .collect(); if reply.send(snap).is_err() { @@ -149,3 +139,74 @@ async fn run_actor(mut rx: mpsc::Receiver, events: broadcast::Sender } } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::SystemTime; + + use super::*; + + fn prefix(s: &str) -> IpPrefix { + IpPrefix::try_from(s).unwrap() + } + + fn peer(port: u16, asn: u32) -> PeerId { + PeerId { + addr: ([192, 0, 2, 1], port).into(), + asn, + } + } + + fn announce(prefixes: &[&str]) -> ReceivedUpdate { + let entry = RibEntry { + path_attributes: Arc::from(vec![]), + next_hop: None, + received_at: SystemTime::now(), + }; + ReceivedUpdate { + withdrawn: vec![], + announced: prefixes + .iter() + .map(|p| (prefix(p), entry.clone())) + .collect(), + } + } + + #[tokio::test] + async fn a_prefix_withdrawn_and_announced_by_one_update_ends_up_announced() { + let rib = AdjRibIn::spawn(); + let a = peer(179, 65001); + rib.apply(a, announce(&["10.0.0.0/8"])).await; + + let mut update = announce(&["10.0.0.0/8"]); + update.withdrawn = vec![prefix("10.0.0.0/8")]; + rib.apply(a, update).await; + + assert_eq!(rib.snapshot().await.len(), 1); + } + + #[tokio::test] + async fn removing_a_peer_leaves_other_peers_routes_alone() { + let rib = AdjRibIn::spawn(); + let (a, b) = (peer(179, 65001), peer(180, 65002)); + rib.apply(a, announce(&["10.0.0.0/8", "10.1.0.0/16"])).await; + rib.apply(b, announce(&["10.0.0.0/8"])).await; + // Commands are handled in order, so once the snapshot is back the + // announcements above have been applied and their events sent. + assert_eq!(rib.snapshot().await.len(), 3); + let mut events = rib.subscribe(); + + assert_eq!(rib.remove_peer(a).await, 2); + + let left = rib.snapshot().await; + assert_eq!(left.len(), 1); + assert_eq!(left[0].peer, b); + for _ in 0..2 { + assert!( + matches!(events.recv().await.unwrap(), RibEvent::Withdrawn { peer, .. } if peer == a) + ); + } + assert_eq!(rib.remove_peer(a).await, 0, "nothing left to remove"); + } +} diff --git a/crates/rib/src/entry.rs b/crates/rib/src/entry.rs index 43d051c..c2b1f7c 100644 --- a/crates/rib/src/entry.rs +++ b/crates/rib/src/entry.rs @@ -1,7 +1,8 @@ use std::net::SocketAddr; +use std::sync::Arc; use std::time::SystemTime; -use bgp_packet::message::PathAttribute; +use bgp_packet::message::{NlriNextHop, PathAttribute, UpdateMessage}; use bgp_primitives::ip_prefix::IpPrefix; /// Identifies a BGP peer by its TCP address and negotiated ASN. @@ -18,21 +19,22 @@ impl std::fmt::Display for PeerId { } /// A single route stored in the Adj-RIB-In. +/// +/// Cloning is cheap: the attribute list is shared, not copied. #[derive(Debug, Clone)] pub struct RibEntry { - pub path_attributes: Vec, + /// The attributes of the route. Every prefix announced by one UPDATE + /// shares one list. MP_REACH_NLRI and MP_UNREACH_NLRI are left out: they + /// carry prefixes, not properties of a route, and the one thing in them + /// that is a property, the next hop, is in `next_hop`. + pub path_attributes: Arc<[PathAttribute]>, + /// Next hop of a route that arrived in MP_REACH_NLRI. `None` for an IPv4 + /// route from the body of the UPDATE, whose next hop is the NEXT_HOP + /// attribute in `path_attributes`. + pub next_hop: Option, pub received_at: SystemTime, } -impl RibEntry { - pub fn new(path_attributes: Vec) -> Self { - Self { - path_attributes, - received_at: SystemTime::now(), - } - } -} - /// A flattened row returned by [`crate::AdjRibIn::snapshot`]. #[derive(Debug, Clone)] pub struct SnapshotEntry { @@ -40,3 +42,156 @@ pub struct SnapshotEntry { pub prefix: IpPrefix, pub entry: RibEntry, } + +/// What one UPDATE message changes in the Adj-RIB-In. +/// +/// An UPDATE names prefixes in four places: withdrawn routes and NLRI in the +/// message body (IPv4 only), and the MP_UNREACH_NLRI and MP_REACH_NLRI +/// attributes (every other family, RFC 4760). This gathers them into the +/// two lists the RIB acts on. +#[derive(Debug, Clone, Default)] +pub struct ReceivedUpdate { + pub withdrawn: Vec, + pub announced: Vec<(IpPrefix, RibEntry)>, +} + +impl ReceivedUpdate { + pub fn from_message(update: &UpdateMessage) -> Self { + Self::from_message_at(update, SystemTime::now()) + } + + fn from_message_at(update: &UpdateMessage, received_at: SystemTime) -> Self { + let mut withdrawn = update.withdrawn_routes.clone(); + // One allocation for the whole UPDATE, however many prefixes it has. + let shared: Arc<[PathAttribute]> = update + .path_attributes + .iter() + .filter(|attr| { + !matches!( + attr, + PathAttribute::MpReachNlri(_) | PathAttribute::MpUnreachNlri(_) + ) + }) + .cloned() + .collect(); + let entry = |next_hop| RibEntry { + path_attributes: Arc::clone(&shared), + next_hop, + received_at, + }; + + let mut announced: Vec<(IpPrefix, RibEntry)> = update + .announced_routes + .iter() + .map(|prefix| (prefix.clone(), entry(None))) + .collect(); + for attr in &update.path_attributes { + match attr { + PathAttribute::MpReachNlri(mp) => announced.extend( + mp.prefixes + .iter() + .map(|prefix| (prefix.clone(), entry(Some(mp.next_hop.clone())))), + ), + PathAttribute::MpUnreachNlri(mp) => withdrawn.extend(mp.prefixes.iter().cloned()), + _ => {} + } + } + Self { + withdrawn, + announced, + } + } +} + +#[cfg(test)] +mod tests { + use std::net::Ipv6Addr; + + use bgp_packet::message::{ + MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, OriginPathAttribute, + }; + use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi}; + + use super::*; + + fn prefix(s: &str) -> IpPrefix { + IpPrefix::try_from(s).unwrap() + } + + fn mp_reach(prefixes: Vec) -> PathAttribute { + PathAttribute::MpReachNlri(MpReachNlriPathAttribute { + afi: AddressFamilyId::Ipv6, + safi: SubsequentAfi::Unicast, + next_hop: NlriNextHop::Ipv6(Ipv6Addr::LOCALHOST), + prefixes, + }) + } + + /// The Adj-RIB-In once kept a full copy of the attribute list per prefix, + /// MP_REACH_NLRI included, so an UPDATE with n prefixes stored n copies + /// of all n prefixes. + #[test] + fn prefixes_of_one_update_share_attributes_that_hold_no_prefixes() { + let prefixes: Vec = (0..500) + .map(|n| prefix(&format!("2001:db8:{n:x}::/48"))) + .collect(); + let update = UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + mp_reach(prefixes.clone()), + ], + announced_routes: vec![], + }; + + let received = ReceivedUpdate::from_message(&update); + + assert_eq!(received.announced.len(), 500); + let first = &received.announced[0].1; + assert_eq!( + first.path_attributes[..], + [PathAttribute::Origin(OriginPathAttribute::IGP)] + ); + for (prefix, entry) in &received.announced { + assert!(prefixes.contains(prefix)); + assert!(Arc::ptr_eq(&entry.path_attributes, &first.path_attributes)); + assert_eq!(entry.next_hop, Some(NlriNextHop::Ipv6(Ipv6Addr::LOCALHOST))); + } + } + + #[test] + fn prefixes_are_gathered_from_the_body_and_the_multiprotocol_attributes() { + let update = UpdateMessage { + withdrawn_routes: vec![prefix("10.0.0.0/8")], + path_attributes: vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + mp_reach(vec![prefix("2001:db8:1::/48")]), + PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute { + afi: AddressFamilyId::Ipv6, + safi: SubsequentAfi::Unicast, + prefixes: vec![prefix("2001:db8:2::/48")], + }), + ], + announced_routes: vec![prefix("192.0.2.0/24")], + }; + + let received = ReceivedUpdate::from_message(&update); + + assert_eq!( + received.withdrawn, + vec![prefix("10.0.0.0/8"), prefix("2001:db8:2::/48")] + ); + let announced: Vec<_> = received + .announced + .iter() + .map(|(p, e)| (p.clone(), e.next_hop.is_some())) + .collect(); + assert_eq!( + announced, + vec![ + (prefix("192.0.2.0/24"), false), + (prefix("2001:db8:1::/48"), true) + ] + ); + } +} diff --git a/crates/rib/src/lib.rs b/crates/rib/src/lib.rs index bbdb277..c64cddd 100644 --- a/crates/rib/src/lib.rs +++ b/crates/rib/src/lib.rs @@ -4,6 +4,6 @@ mod event; mod local_rib; pub use adj_rib_in::AdjRibIn; -pub use entry::{PeerId, RibEntry, SnapshotEntry}; +pub use entry::{PeerId, ReceivedUpdate, RibEntry, SnapshotEntry}; pub use event::RibEvent; pub use local_rib::{LocalRib, LocalRibError, LocalRibEvent, LocalRoute, LocalRouteAttrs}; diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index 42e09ea..3887a48 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -6,7 +6,7 @@ use bgp_packet::message::{BGP_HEADER_LEN, Message, NotificationMessage, PathAttr use bgp_packet::open::{Capability, OpenMessage, OpenOption}; use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi}; use bgp_primitives::parser::ParserContext; -use bgp_rib::{AdjRibIn, PeerId, RibEntry}; +use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate}; use bytes::BytesMut; use eyre::{Result, bail, eyre}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -290,8 +290,7 @@ async fn established_loop( hold_timer.as_mut().reset(Instant::now() + hold_duration); let (withdrawn, announced) = prefix_counts(&update); info!(peer = %peer_id, withdrawn, announced, "Received UPDATE"); - apply_update(adj_rib_in, peer_id, &update.path_attributes, - &update.withdrawn_routes, &update.announced_routes).await; + adj_rib_in.apply(peer_id, ReceivedUpdate::from_message(&update)).await; } Message::Notification(notif) => { info!(peer = %peer_id, error = %notif.error, "Received NOTIFICATION, closing session"); @@ -434,53 +433,6 @@ fn prefix_counts(update: &bgp_packet::message::UpdateMessage) -> (usize, usize) (withdrawn, announced) } -/// Applies a received UPDATE to the Adj-RIB-In. -/// -/// Handles both the legacy IPv4 NLRI fields and the MP_REACH/MP_UNREACH path -/// attributes used for all other address families (RFC 2858). -async fn apply_update( - rib: &AdjRibIn, - peer: PeerId, - path_attributes: &[PathAttribute], - withdrawn: &[bgp_primitives::ip_prefix::IpPrefix], - announced: &[bgp_primitives::ip_prefix::IpPrefix], -) { - // Legacy IPv4 fields in the UPDATE body. - for prefix in withdrawn { - rib.withdraw(peer, prefix.clone()).await; - } - for prefix in announced { - rib.announce( - peer, - prefix.clone(), - RibEntry::new(path_attributes.to_vec()), - ) - .await; - } - - // Multiprotocol attributes carry NLRI for all other AFIs. - for attr in path_attributes { - match attr { - PathAttribute::MpReachNlri(mp) => { - for prefix in &mp.prefixes { - rib.announce( - peer, - prefix.clone(), - RibEntry::new(path_attributes.to_vec()), - ) - .await; - } - } - PathAttribute::MpUnreachNlri(mp) => { - for prefix in &mp.prefixes { - rib.withdraw(peer, prefix.clone()).await; - } - } - _ => {} - } - } -} - // --------------------------------------------------------------------------- // OPEN construction and validation // --------------------------------------------------------------------------- diff --git a/crates/web/src/api.rs b/crates/web/src/api.rs index 82eaf29..a405f2f 100644 --- a/crates/web/src/api.rs +++ b/crates/web/src/api.rs @@ -7,7 +7,7 @@ use axum::Json; use axum::extract::State; use bgp_packet::community; use bgp_packet::message::{OriginPathAttribute, PathAttribute}; -use bgp_rib::{LocalRoute, SnapshotEntry}; +use bgp_rib::{LocalRoute, RibEntry, SnapshotEntry}; use bgp_server::session::{ConfiguredPeer, DialState, SessionHandle, SessionRegistry}; use serde::Serialize; @@ -181,9 +181,14 @@ pub struct AttrSummary { pub large_communities: Vec, } -pub fn summarize(attrs: &[PathAttribute]) -> AttrSummary { - let mut s = AttrSummary::default(); - for attr in attrs { +pub fn summarize(entry: &RibEntry) -> AttrSummary { + let mut s = AttrSummary { + // A multiprotocol route has its next hop here; an IPv4 route has it + // in the NEXT_HOP attribute below. + next_hop: entry.next_hop.as_ref().map(ToString::to_string), + ..Default::default() + }; + for attr in entry.path_attributes.iter() { match attr { PathAttribute::Origin(o) => s.origin = Some(origin_name(*o).to_string()), PathAttribute::AsPath(p) => { @@ -194,7 +199,6 @@ pub fn summarize(attrs: &[PathAttribute]) -> AttrSummary { .collect() } PathAttribute::NextHop(nh) => s.next_hop = Some(nh.0.to_string()), - PathAttribute::MpReachNlri(mp) => s.next_hop = Some(format!("{:?}", mp.next_hop)), PathAttribute::MultiExitDisc(m) => s.med = Some(m.0), PathAttribute::LocalPref(l) => s.local_pref = Some(l.0), PathAttribute::Communities(c) => { @@ -220,7 +224,8 @@ pub struct ReceivedRouteView { pub prefix: String, pub received_at_unix: u64, pub summary: AttrSummary, - /// The full decoded attribute list, for the curious. + /// The decoded attribute list. MP_REACH_NLRI is not in it: its next hop + /// is in the summary and its prefixes are routes of their own. pub attributes: Vec, } @@ -231,8 +236,8 @@ impl From<&SnapshotEntry> for ReceivedRouteView { peer_asn: e.peer.asn, prefix: e.prefix.to_string(), received_at_unix: unix_secs(e.entry.received_at), - summary: summarize(&e.entry.path_attributes), - attributes: e.entry.path_attributes.clone(), + summary: summarize(&e.entry), + attributes: e.entry.path_attributes.to_vec(), } } } diff --git a/crates/web/src/events.rs b/crates/web/src/events.rs index 046d5ba..c1ec555 100644 --- a/crates/web/src/events.rs +++ b/crates/web/src/events.rs @@ -102,8 +102,8 @@ impl WireEvent { .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_secs()) .unwrap_or(0), - summary: summarize(&entry.path_attributes), - attributes: entry.path_attributes, + summary: summarize(&entry), + attributes: entry.path_attributes.to_vec(), }, }, Ok(RibEvent::Withdrawn { peer, prefix }) => Self::ReceivedWithdrawn {