export: one rule each for admission, identifiers and what the peer had

A code-quality pass over the exporter. The resync path (gather then
stream) had grown beside the steady-state path (judge then flush)
with its own copies of the admission predicate, the path identifier
rule, the reconciliation of a route the peer already had, the parser
contexts, the withdrawal loop and the advertised gauge. Each is one
function now: `prefix_may_go`, `lowest_free_id`, `Sent::renewed`,
`parser_contexts`, `send_withdrawals`, `within_limit` and
`note_held`. The 3,072-octet ceiling on a route's attributes, which
is what keeps an UPDATE within its size, is a named constant where it
is relied on. The UPDATEs are framed with the packet crate's helpers
instead of a bare type code and a length written by hand. The
positional tuples for a prefix's paths are structs with names. A
mismatched kind of bookkeeping, which nothing should produce, is
asserted in tests and logged in release instead of dropped silently.
Types used only inside the crate are `pub(crate)`. Five doc comments
that had drifted onto neighbours or named former designs are put
right; the exporter's reference encoder no longer breaks the CI doc
build with a link to an item that exists only under test.

Testing: ADD-PATH export has unit tests at last: two routes for a
prefix go with their own identifiers and keep them, and the
withdrawal of one carries its identifier; what is not heard of again
after a reset is withdrawn after the table. The byte-for-byte test of
the encoded UPDATEs covers IPv4 routes with an IPv6 next hop (RFC
8950) too. The test fixtures are shared helpers. The server tests and
the ADD-PATH, transit, limits, roles and deadlock integration tests
pass.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Rayhaan JaufeerallyandClaude Fable 5.1 committed 2026-10-10 22:44:32 +02:00
1 parent bef875e50b
commit 4cddd6fb04
3 files changed
+698 -489

No files matched your search

File diff suppressed because it is too large. Load diff
+130 -108
View File
@@ -33,9 +33,9 @@
use std::net::{IpAddr, Ipv4Addr}; use std::net::{IpAddr, Ipv4Addr};
use bgp_packet::message::{ use bgp_packet::message::{
AsPathAttribute, AsPathSegment, BGP_MARKER, LocalPrefPathAttribute, MpReachNlriPathAttribute, AsPathAttribute, AsPathSegment, LocalPrefPathAttribute, MpReachNlriPathAttribute,
MpUnreachNlriPathAttribute, NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage, MpUnreachNlriPathAttribute, NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage,
attr_type, write_nlri, attr_type, begin_message, finish_message, msg_type, put_nlri_list,
}; };
use bgp_primitives::constants::{AfiSafi, IpVersion}; use bgp_primitives::constants::{AfiSafi, IpVersion};
use bgp_primitives::ip_prefix::Nlri; use bgp_primitives::ip_prefix::Nlri;
@@ -50,7 +50,7 @@ use crate::policy::{ExportPolicy, PeerRole};
/// The session a route may go out on, as far as that decides whether it /// The session a route may go out on, as far as that decides whether it
/// does and as what. /// does and as what.
#[derive(Debug, Clone, Copy)] #[derive(Debug, Clone, Copy)]
pub struct Egress<'a> { pub(crate) struct Egress<'a> {
pub towards: Towards, pub towards: Towards,
pub peer_asn: u32, pub peer_asn: u32,
/// The peer's number in the routing table: what its own routes there /// The peer's number in the routing table: what its own routes there
@@ -69,7 +69,7 @@ pub struct Egress<'a> {
/// Route reflection (RFC 4456), as far as one session goes. /// Route reflection (RFC 4456), as far as one session goes.
#[derive(Debug, Clone, Copy)] #[derive(Debug, Clone, Copy)]
pub struct Reflection { pub(crate) struct Reflection {
/// The cluster this router reflects routes for. /// The cluster this router reflects routes for.
pub cluster_id: Ipv4Addr, pub cluster_id: Ipv4Addr,
/// Whether the peer is a route reflector client of ours. /// Whether the peer is a route reflector client of ours.
@@ -81,7 +81,7 @@ pub struct Reflection {
/// Where a learned route came from. /// Where a learned route came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Source { pub(crate) struct Source {
pub peer: PeerIndex, pub peer: PeerIndex,
/// Whether that peer is in our own AS. /// Whether that peer is in our own AS.
pub ibgp: bool, pub ibgp: bool,
@@ -95,7 +95,7 @@ pub struct Source {
/// A route ready to be announced on a session: its attributes, but for /// A route ready to be announced on a session: its attributes, but for
/// the ones that depend on which prefixes go with them. /// the ones that depend on which prefixes go with them.
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct Outgoing { pub(crate) struct Outgoing {
pub attributes: Vec<PathAttribute>, pub attributes: Vec<PathAttribute>,
pub next_hop: NlriNextHop, pub next_hop: NlriNextHop,
} }
@@ -396,6 +396,14 @@ pub(crate) fn finish(attributes: &mut Vec<PathAttribute>, towards: &Towards) {
/// lengths every UPDATE has. /// lengths every UPDATE has.
const UPDATE_ROOM: usize = 4096 - 19 - 4; const UPDATE_ROOM: usize = 4096 - 19 - 4;
/// The most octets a route's attributes may take on the wire for the
/// route to go out at all. A message is 4,096 octets at most (RFC 4271
/// §4.1) and must hold the attributes, the next hop and at least one
/// prefix; three quarters of it leaves [`UPDATE_ROOM`] a thousand octets
/// for the next hop and the prefixes, which is some forty IPv6 prefixes
/// with path identifiers a message, enough to keep a table's UPDATEs few.
pub(crate) const ATTRIBUTES_AT_MOST: usize = 3072;
/// The most octets a prefix takes in an UPDATE: a path identifier, a /// The most octets a prefix takes in an UPDATE: a path identifier, a
/// length, and the address. /// length, and the address.
fn nlri_octets(routes: IpVersion) -> usize { fn nlri_octets(routes: IpVersion) -> usize {
@@ -409,16 +417,16 @@ fn nlri_octets(routes: IpVersion) -> usize {
/// MP_REACH_NLRI (type 14) go, so that the attributes stay in order of /// MP_REACH_NLRI (type 14) go, so that the attributes stay in order of
/// type code. /// type code.
#[derive(Debug)] #[derive(Debug)]
pub struct EncodedAttributes { pub(crate) struct EncodedAttributes {
pub bytes: Bytes, bytes: Bytes,
pub at_next_hop: usize, at_next_hop: usize,
pub at_mp_reach: usize, at_mp_reach: usize,
} }
impl EncodedAttributes { impl EncodedAttributes {
/// Encodes `attributes`, which are `finish`ed, for a session with /// Encodes `attributes`, which are `finish`ed, for a session with
/// `four_octet_asn`. `None` if they cannot be encoded. /// `four_octet_asn`. `None` if they cannot be encoded.
pub fn encode(attributes: &[PathAttribute], four_octet_asn: bool) -> Option<Self> { pub(crate) fn encode(attributes: &[PathAttribute], four_octet_asn: bool) -> Option<Self> {
let ctx = ParserContext { let ctx = ParserContext {
four_octet_asn, four_octet_asn,
ibgp: false, ibgp: false,
@@ -444,9 +452,81 @@ impl EncodedAttributes {
}) })
} }
pub fn octets(&self) -> usize { pub(crate) fn octets(&self) -> usize {
self.bytes.len() self.bytes.len()
} }
/// The UPDATE messages, complete with their headers, that announce
/// `routes`, all of IP version `version`, with these attributes and
/// `next_hop`: as few as they fit in, in pieces of [`ENCODED_PIECE`]
/// octets or so, each of whole messages. `ctx` is the session's, and
/// says whether path identifiers go with the family's prefixes
/// (RFC 7911).
///
/// What `announcements`, kept under test as the reference, builds as
/// messages, built as bytes: the attributes are not parsed, cloned or
/// encoded again for each message.
pub(crate) fn announcements(
&self,
next_hop: NlriNextHop,
version: IpVersion,
routes: &[Nlri],
ctx: &ParserContext,
) -> Vec<BytesMut> {
// Attributes that would leave no room for a prefix do not go out
// at all; see `Learned::dressed`.
debug_assert!(self.octets() < ATTRIBUTES_AT_MOST);
let room = UPDATE_ROOM.saturating_sub(self.octets() + NEXT_HOP_ROOM);
let at_once = (room / nlri_octets(version)).max(1);
let in_body = matches!(next_hop, NlriNextHop::Ipv4(_)) && version == IpVersion::Ipv4;
let with_path_id = ctx.add_path.contains(&AfiSafi::unicast(version));
let mut pieces = Vec::new();
let mut out = BytesMut::with_capacity(ENCODED_PIECE.min(routes.len() * 64 + 4096));
for routes in routes.chunks(at_once) {
{
let length_at = begin_message(&mut out, msg_type::UPDATE);
// No withdrawn routes.
out.put_u16(0);
let attributes_at = out.len();
out.put_u16(0);
if in_body {
let NlriNextHop::Ipv4(next_hop) = next_hop else {
unreachable!("in_body means an IPv4 next hop");
};
out.put_slice(&self.bytes[..self.at_next_hop]);
PathAttribute::NextHop(NextHopPathAttribute(next_hop))
.to_wire(ctx, &mut out)
.expect("a next hop encodes");
out.put_slice(&self.bytes[self.at_next_hop..]);
} else {
out.put_slice(&self.bytes[..self.at_mp_reach]);
PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
family: AfiSafi::unicast(version),
next_hop,
prefixes: routes.to_vec(),
})
.to_wire(ctx, &mut out)
.expect("a chunk that fits encodes");
out.put_slice(&self.bytes[self.at_mp_reach..]);
}
let attributes_len = out.len() - attributes_at - 2;
out[attributes_at..attributes_at + 2]
.copy_from_slice(&(attributes_len as u16).to_be_bytes());
if in_body {
put_nlri_list(routes, with_path_id, &mut out);
}
// As many prefixes went in as fit beside the attributes.
finish_message(&mut out, length_at).expect("a chunk that fits is a message");
}
if out.len() >= ENCODED_PIECE {
pieces.push(out.split());
}
}
if !out.is_empty() {
pieces.push(out);
}
pieces
}
} }
/// What the next hop takes, in its own attribute or with the prefixes, /// What the next hop takes, in its own attribute or with the prefixes,
@@ -458,84 +538,12 @@ const NEXT_HOP_ROOM: usize = 64;
/// so it runs over by at most one. /// so it runs over by at most one.
pub(crate) const ENCODED_PIECE: usize = 64 * 1024; pub(crate) const ENCODED_PIECE: usize = 64 * 1024;
/// The UPDATE messages, complete with their headers, that announce
/// `routes`, all of IP version `version`, with the attributes `encoded`
/// and `next_hop`: as few as they fit in, in pieces of
/// [`ENCODED_PIECE`] octets or so, each of whole messages. `ctx` is the
/// session's, and says whether path identifiers go with the family's
/// prefixes (RFC 7911).
///
/// What [`announcements`] builds as messages, built as bytes: the
/// attributes are not parsed, cloned or encoded again for each message.
pub(crate) fn encoded_announcements(
encoded: &EncodedAttributes,
next_hop: NlriNextHop,
version: IpVersion,
routes: &[Nlri],
ctx: &ParserContext,
) -> Vec<BytesMut> {
let room = UPDATE_ROOM.saturating_sub(encoded.octets() + NEXT_HOP_ROOM);
let at_once = (room / nlri_octets(version)).max(1);
let in_body = matches!(next_hop, NlriNextHop::Ipv4(_)) && version == IpVersion::Ipv4;
let with_path_id = ctx.add_path.contains(&AfiSafi::unicast(version));
let mut pieces = Vec::new();
let mut out = BytesMut::with_capacity(ENCODED_PIECE.min(routes.len() * 64 + 4096));
for routes in routes.chunks(at_once) {
{
out.put_slice(&BGP_MARKER);
let length_at = out.len();
out.put_u16(0);
// The type of an UPDATE (RFC 4271 §4.1).
out.put_u8(2);
// No withdrawn routes.
out.put_u16(0);
let attributes_at = out.len();
out.put_u16(0);
if in_body {
let NlriNextHop::Ipv4(next_hop) = next_hop else {
unreachable!("in_body means an IPv4 next hop");
};
out.put_slice(&encoded.bytes[..encoded.at_next_hop]);
PathAttribute::NextHop(NextHopPathAttribute(next_hop))
.to_wire(ctx, &mut out)
.expect("a next hop encodes");
out.put_slice(&encoded.bytes[encoded.at_next_hop..]);
} else {
out.put_slice(&encoded.bytes[..encoded.at_mp_reach]);
PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
family: AfiSafi::unicast(version),
next_hop,
prefixes: routes.to_vec(),
})
.to_wire(ctx, &mut out)
.expect("a chunk that fits encodes");
out.put_slice(&encoded.bytes[encoded.at_mp_reach..]);
}
let attributes_len = out.len() - attributes_at - 2;
out[attributes_at..attributes_at + 2]
.copy_from_slice(&(attributes_len as u16).to_be_bytes());
if in_body {
write_nlri(routes, with_path_id, &mut out);
}
let length = out.len() - length_at + BGP_MARKER.len();
out[length_at..length_at + 2].copy_from_slice(&(length as u16).to_be_bytes());
}
if out.len() >= ENCODED_PIECE {
pieces.push(out.split());
}
}
if !out.is_empty() {
pieces.push(out);
}
pieces
}
/// The UPDATEs that announce `routes`, all of IP version `version`, as /// The UPDATEs that announce `routes`, all of IP version `version`, as
/// `outgoing`: as few as they fit in. `attributes_octets` is how long /// `outgoing`: as few as they fit in. `attributes_octets` is how long
/// `outgoing`'s attributes are on this session's wire. /// `outgoing`'s attributes are on this session's wire.
/// ///
/// What [`encoded_announcements`] does, built as messages: kept as what /// What [`EncodedAttributes::announcements`] does, built as messages:
/// its bytes are checked against. /// kept as what its bytes are checked against.
#[cfg(test)] #[cfg(test)]
pub(crate) fn announcements( pub(crate) fn announcements(
outgoing: &Outgoing, outgoing: &Outgoing,
@@ -544,9 +552,6 @@ pub(crate) fn announcements(
towards: &Towards, towards: &Towards,
attributes_octets: usize, attributes_octets: usize,
) -> Vec<UpdateMessage> { ) -> Vec<UpdateMessage> {
// What the next hop takes, in its own attribute or with the prefixes,
// and a little to spare.
const NEXT_HOP_ROOM: usize = 64;
let room = UPDATE_ROOM.saturating_sub(attributes_octets + NEXT_HOP_ROOM); let room = UPDATE_ROOM.saturating_sub(attributes_octets + NEXT_HOP_ROOM);
let at_once = (room / nlri_octets(version)).max(1); let at_once = (room / nlri_octets(version)).max(1);
let in_body = matches!(outgoing.next_hop, NlriNextHop::Ipv4(_)) && version == IpVersion::Ipv4; let in_body = matches!(outgoing.next_hop, NlriNextHop::Ipv4(_)) && version == IpVersion::Ipv4;
@@ -684,30 +689,46 @@ mod tests {
PathAttribute::Communities(CommunitiesPathAttribute(communities.to_vec())) PathAttribute::Communities(CommunitiesPathAttribute(communities.to_vec()))
} }
/// What `encoded_announcements` writes is, byte for byte, what the /// What `EncodedAttributes::announcements` writes is, byte for byte,
/// messages `announcements` builds encode to: with the next hop in /// what the messages `announcements` builds encode to: with the next
/// its own attribute (IPv4) and in MP_REACH_NLRI (IPv6), with and /// hop in its own attribute (IPv4) and in MP_REACH_NLRI (IPv6, and
/// without path identifiers, to two-octet and four-octet AS speakers /// IPv4 with an IPv6 next hop, RFC 8950), with and without path
/// (so with the attributes around NEXT_HOP, MP_REACH_NLRI and /// identifiers, to two-octet and four-octet AS speakers (so with the
/// AS4_PATH in order), and over as many messages when the routes do /// attributes around NEXT_HOP, MP_REACH_NLRI and AS4_PATH in order),
/// not fit in one. /// and over as many messages when the routes do not fit in one.
#[test] #[test]
fn encoded_announcements_are_the_messages_on_the_wire() { fn encoded_announcements_are_the_messages_on_the_wire() {
use std::net::Ipv6Addr; use std::net::Ipv6Addr;
for four_octet_asn in [true, false] { // The routes' IP version, their next hop, and whether the session
for with_path_id in [false, true] { // takes IPv4 routes with IPv6 next hops.
for version in [IpVersion::Ipv4, IpVersion::Ipv6] { let layouts = [
let towards = Towards { (
four_octet_asn, IpVersion::Ipv4,
..towards(None, false) NlriNextHop::Ipv4(Ipv4Addr::new(192, 0, 2, 1)),
}; false,
let next_hop = match version { ),
IpVersion::Ipv4 => NlriNextHop::Ipv4(Ipv4Addr::new(192, 0, 2, 1)), (
IpVersion::Ipv6 => NlriNextHop::Ipv6WithLl { IpVersion::Ipv6,
NlriNextHop::Ipv6WithLl {
global: Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1), global: Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1),
link_local: Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 1), link_local: Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 1),
}, },
false,
),
(
IpVersion::Ipv4,
NlriNextHop::Ipv6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1)),
true,
),
];
for four_octet_asn in [true, false] {
for with_path_id in [false, true] {
for (version, next_hop, extended_next_hop) in layouts {
let towards = Towards {
four_octet_asn,
extended_next_hop,
..towards(None, false)
}; };
let attributes = vec![ let attributes = vec![
PathAttribute::Origin(OriginPathAttribute::IGP), PathAttribute::Origin(OriginPathAttribute::IGP),
@@ -753,7 +774,7 @@ mod tests {
let mut finished = attributes.clone(); let mut finished = attributes.clone();
finish(&mut finished, &towards); finish(&mut finished, &towards);
let encoded = EncodedAttributes::encode(&finished, four_octet_asn).unwrap(); let encoded = EncodedAttributes::encode(&finished, four_octet_asn).unwrap();
let bytes = encoded_announcements(&encoded, next_hop, version, &routes, &ctx); let bytes = encoded.announcements(next_hop, version, &routes, &ctx);
let outgoing = Outgoing { let outgoing = Outgoing {
attributes, attributes,
@@ -771,7 +792,8 @@ mod tests {
.collect(); .collect();
let what = format!( let what = format!(
"{version:?}, path ids {with_path_id}, four-octet {four_octet_asn}" "{version:?} with next hop {next_hop:?}, path ids {with_path_id}, \
four-octet {four_octet_asn}"
); );
assert!(expected.len() > 1, "{what}: more than one message"); assert!(expected.len() > 1, "{what}: more than one message");
for update in &expected { for update in &expected {
+2 -6
View File
@@ -295,8 +295,6 @@ impl SessionHandle {
} }
} }
/// Whether `other` is a handle to this same session, and not to another
/// session with the same peer.
/// What the peer goes by in the metrics: its configured name, or its /// What the peer goes by in the metrics: its configured name, or its
/// address. /// address.
pub fn metric_label(&self) -> String { pub fn metric_label(&self) -> String {
@@ -305,6 +303,8 @@ impl SessionHandle {
.unwrap_or_else(|| crate::address::PeerAddress::from(self.peer.addr).to_string()) .unwrap_or_else(|| crate::address::PeerAddress::from(self.peer.addr).to_string())
} }
/// Whether `other` is a handle to this same session, and not to another
/// session with the same peer.
pub fn is_same_session(&self, other: &SessionHandle) -> bool { pub fn is_same_session(&self, other: &SessionHandle) -> bool {
self.outbound.same_channel(&other.outbound) self.outbound.same_channel(&other.outbound)
} }
@@ -460,10 +460,6 @@ impl SessionRegistry {
self.shutting_down.load(Ordering::Relaxed) self.shutting_down.load(Ordering::Relaxed)
} }
/// Closes every session politely: each peer is sent Cease /
/// Administrative Shutdown (RFC 4486) so it can drop the session at once
/// instead of discovering a dead TCP connection. Waits up to `grace` for
/// the sessions to finish; returns how many were still open.
/// Closes every connection because the daemon is going, as `leave` /// Closes every connection because the daemon is going, as `leave`
/// says: for good, or to restart (see [`Leave`]). Waits up to `grace` /// says: for good, or to restart (see [`Leave`]). Waits up to `grace`
/// for the sessions to finish; returns how many were still open. /// for the sessions to finish; returns how many were still open.