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 <noreply@anthropic.com>

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 {

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<RibCommand>,
@@ -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<RibCommand>, events: broadcast::Sender<RibEvent>) {
-    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<PeerId, HashMap<IpPrefix, RibEntry>> = 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<IpPrefix> = 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<RibCommand>, 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");
+    }
+}

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<PathAttribute>,
+    /// 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<NlriNextHop>,
     pub received_at: SystemTime,
 }
 
-impl RibEntry {
-    pub fn new(path_attributes: Vec<PathAttribute>) -> 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<IpPrefix>,
+    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<IpPrefix>) -> 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<IpPrefix> = (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)
+            ]
+        );
+    }
+}

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};

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
 // ---------------------------------------------------------------------------

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<String>,
 }
 
-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<PathAttribute>,
 }
 
@@ -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(),
         }
     }
 }

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 {