diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index b5e6fff..95af20f 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -111,7 +111,7 @@ struct Advertiser { /// 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`. + /// see `Learned::gathering`. resyncs: Arc, session_events: Receiver, exporters: HashMap, @@ -161,7 +161,7 @@ impl Advertiser { // 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) + Learned::new(reader_of(&self.rib, all_paths), all_paths) }), rib: Arc::clone(&self.rib), resyncs: Arc::clone(&self.resyncs), @@ -188,16 +188,6 @@ impl Advertiser { 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. @@ -217,6 +207,16 @@ impl Advertiser { } } +/// A new follower of `rib` for an exporter: of every route, or of the +/// best ones. Either starts with the whole table. +fn reader_of(rib: &Arc, all_paths: bool) -> Reader { + if all_paths { + rib.reader_of_all_routes() + } else { + rib.reader() + } +} + /// 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 { @@ -285,14 +285,25 @@ enum News { }, 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. + /// route. Routes { prefix: IpPrefix, - routes: Vec<(PeerIndex, u32, Arc, bool)>, + routes: Vec, }, } +/// One of a prefix's routes, as the table has it. +struct Path { + /// Who it is from. + peer: PeerIndex, + /// That peer's identifier for it: zero unless the peer sends several + /// routes for a prefix (ADD-PATH, RFC 7911). + path_id: u32, + attrs: Arc, + /// Whether it is the best of the prefix's routes. + best: 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); @@ -324,15 +335,46 @@ struct Sent { attrs: Arc, } +impl Sent { + /// The route as the peer is to hold it now that the table has it with + /// `attrs`, and whether that is to go out: not if `attrs` are the set + /// it went out with, since what it goes out as follows from them; and + /// under the identifier that is its own if they are not. + fn renewed(mut self, attrs: &Arc) -> (Self, bool) { + if Arc::ptr_eq(&self.attrs, attrs) { + return (self, false); + } + self.attrs = Arc::clone(attrs); + (self, true) + } +} + +/// The lowest path identifier (ADD-PATH, RFC 7911) not among `in_use`: +/// what a route of a prefix that the peer has not had goes out under. +/// Identifiers start at one; zero is a route not told apart. +fn lowest_free_id(in_use: impl Iterator + Clone) -> u32 { + (1..) + .find(|id| !in_use.clone().any(|used| used == *id)) + .expect("fewer routes than identifiers") +} + /// 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>), - /// 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 + /// Every route that goes out; empty for a prefix the peer is not to /// have. - All(Vec<(PathKey, Arc, Arc)>), + All(Vec), +} + +/// A route the peer is to have, for a session that is told of every +/// route: what tells it apart, the attributes it came with, and what it +/// goes out as. +struct Wanted { + key: PathKey, + attrs: Arc, + dressed: Arc, } /// What the peer is to end up with for each prefix heard of. @@ -437,8 +479,16 @@ impl Announced { routes.insert(prefix, sent); } } - // Both are of one session, which is one or the other. - _ => {} + // Both are of one session, which is one or the other. Were + // they not, what `other` holds would stay with the peer + // unaccounted for; better said than lost in silence. + (_, other) => { + debug_assert!(false, "the learned routes of one session are kept one way"); + warn!( + routes = other.routes(), + "Routes announced to a peer were kept the wrong way and are not accounted for" + ); + } } } @@ -542,13 +592,11 @@ impl Learned { prefix, routes: routes .iter() - .map(|route| { - ( - route.peer, - route.path_id, - Arc::clone(route.attrs), - route.best, - ) + .map(|route| Path { + peer: route.peer, + path_id: route.path_id, + attrs: Arc::clone(route.attrs), + best: route.best, }) .collect(), }, @@ -621,7 +669,7 @@ impl Learned { })) }) // One that cannot be written is one that does not go. - .filter(|dressed| dressed.octets < 3072); + .filter(|dressed| dressed.octets < export::ATTRIBUTES_AT_MOST); if self.dressed.len() >= DRESSED_KEPT { self.dressed.clear(); } @@ -710,11 +758,7 @@ impl Exporter { // 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() - }; + learned.reader = reader_of(&self.rib, learned.all_paths); } } () = changed(&mut self.learned) => { @@ -761,7 +805,8 @@ impl Exporter { } /// What every route on its way to `session` goes through. Not a method: - /// `follow` holds the learned routes mutably while it uses this. + /// `judge` and `stream` hold the learned routes mutably while they use + /// this; `announce` has no such need, and uses it all the same. fn egress( session: &SessionHandle, control: Option, @@ -786,10 +831,9 @@ impl Exporter { /// Notes how many routes the peer now holds from us, for the status /// page and the metrics. fn note_advertised(&self) { - let advertised = + let held = 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); + note_held(&self.session, &self.advertised_metric, held); } /// The session as far as it decides what goes out on it and how. @@ -805,8 +849,6 @@ impl Exporter { } } - /// Takes the next changes of best route from the table and tells the - /// peer of those that are for it. /// Whether the last look at the table saw everything in it. fn caught_up(&self) -> bool { self.learned @@ -814,6 +856,8 @@ impl Exporter { .is_none_or(|learned| learned.reader.caught_up()) } + /// 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(()); @@ -894,14 +938,6 @@ impl Exporter { else { return; }; - let may_go = |prefix: &IpPrefix| { - session.families.carries(prefix) - && !self.own_announced.contains(prefix) - && session - .export_limits - .as_deref() - .is_none_or(|limits| limits.allows(prefix)) - }; match item { // Again, before it was all there: from the start. News::Reset => *gathered = Gathered::default(), @@ -913,8 +949,11 @@ impl Exporter { } => { // The peer has gone, and its routes with it. if let Some(source) = source_of(&self.rib, &mut gathered.sources, peer) - && may_go(&prefix) + && prefix_may_go(session, &self.own_announced, &prefix) { + // A session that is sent only the best route is not + // told the routes apart: the one route of the prefix + // goes by the peer it is from, under no identifier. let entry = Entry { prefix, key: (peer, 0), @@ -925,7 +964,7 @@ impl Exporter { } News::Routes { prefix, routes } => { let version = prefix.address_family(); - if !may_go(&prefix) { + if !prefix_may_go(session, &self.own_announced, &prefix) { return; } // Where the peer did not agree to tell the routes of a @@ -933,10 +972,10 @@ impl Exporter { let told_apart = session.add_path_send.contains(AfiSafi::unicast(version)); let wanted: Vec<(PathKey, Arc, Source)> = routes .into_iter() - .filter(|(_, _, _, best)| told_apart || *best) - .filter_map(|(peer, path_id, attrs, _)| { - source_of(&self.rib, &mut gathered.sources, peer) - .map(|source| ((peer, path_id), attrs, source)) + .filter(|route| told_apart || route.best) + .filter_map(|route| { + source_of(&self.rib, &mut gathered.sources, route.peer) + .map(|source| ((route.peer, route.path_id), route.attrs, source)) }) .collect(); // A route the peer had keeps its identifier; a new one @@ -952,11 +991,9 @@ impl Exporter { for (key, attrs, source) in wanted { let id = match had.iter().find(|sent| sent.key == key) { Some(sent) => sent.id, - None => (1..) - .find(|id| { - !had.iter().any(|sent| sent.id == *id) && !taken.contains(id) - }) - .expect("fewer routes than identifiers"), + None => lowest_free_id( + had.iter().map(|sent| sent.id).chain(taken.iter().copied()), + ), }; taken.push(id); gathered.add(&attrs, source, version, Entry { prefix, key, id }); @@ -989,13 +1026,8 @@ impl Exporter { // that is more than it may. Gathered routes are at most that // many; only if those are over the limit is each set dressed // first, to know which cannot be and take their routes with them. - let limits = session.export_limits.as_deref(); let at_most: usize = groups.values().map(|group| group.routes.len()).sum(); - if limits.is_some_and(|limits| { - limits - .exceeded_by(self.own_announced.len() + at_most) - .is_some() - }) { + if exceeded(session, self.own_announced.len() + at_most).is_some() { let mut will_hold = 0; for (n, ((_, _, version), group)) in groups.iter().enumerate() { if n % EXPORT_BATCH == EXPORT_BATCH - 1 { @@ -1008,20 +1040,9 @@ impl Exporter { will_hold += group.routes.len(); } } - let routes = self.own_announced.len() + will_hold; - if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) { - return Err(over_limit(session, limit, routes)); - } + within_limit(session, self.own_announced.len() + will_hold)?; } - let contexts = [IpVersion::Ipv4, IpVersion::Ipv6].map(|version| ParserContext { - four_octet_asn: towards.four_octet_asn, - ibgp: towards.ibgp, - add_path: if session.add_path_send.contains(AfiSafi::unicast(version)) { - vec![AfiSafi::unicast(version)] - } else { - Vec::new() - }, - }); + let contexts = parser_contexts(session, &towards); let mut entries = 0; for ((_, _, version), group) in groups { // Dressed as its turn comes: the peer is reading the sets @@ -1065,21 +1086,26 @@ impl Exporter { } let has = routes.entry(entry.prefix).or_default(); *count += 1; - if let Some(sent) = known - && Arc::ptr_eq(&sent.attrs, &attrs) - { - has.push(sent); - continue; + // Gathered under the identifier it had, if it had + // one; so a route kept keeps the identifier too. + let (sent, goes) = match known { + Some(sent) => sent.renewed(&attrs), + None => ( + Sent { + key: entry.key, + id: entry.id, + attrs: Arc::clone(&attrs), + }, + true, + ), + }; + if goes { + nlris.push(Nlri { + prefix: entry.prefix, + path_id: sent.id, + }); } - has.push(Sent { - key: entry.key, - id: entry.id, - attrs: Arc::clone(&attrs), - }); - nlris.push(Nlri { - prefix: entry.prefix, - path_id: entry.id, - }); + has.push(sent); } } } @@ -1089,8 +1115,7 @@ impl Exporter { ROUTES_EXPORTED .with_label_values(&["announced"]) .inc_by(nlris.len() as u64); - let updates = export::encoded_announcements( - &dressed.encoded, + let updates = dressed.encoded.announcements( dressed.next_hop, version, &nlris, @@ -1099,27 +1124,16 @@ impl Exporter { for update in updates { session.send_encoded(update).await?; } - let advertised = self.own_announced.len() + learned.held_by_peer(); - session.counters.set_advertised(advertised); - self.advertised_metric.set(advertised as i64); + note_held( + session, + &self.advertised_metric, + self.own_announced.len() + learned.held_by_peer(), + ); } // Not heard of again: gone. learned.synced = true; if let Some(stale) = learned.stale.take() { - for (version, routes) in [IpVersion::Ipv4, IpVersion::Ipv6] - .into_iter() - .zip(stale.withdrawals()) - { - 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?; - } - } + send_withdrawals(session, &towards, stale.withdrawals()).await?; } self.note_advertised(); Ok(()) @@ -1134,15 +1148,10 @@ impl Exporter { let Some(learned) = self.learned.as_mut() else { return verdicts; }; - let limits = session.export_limits.as_deref(); let egress = Self::egress(session, self.control, self.cluster_id, towards); // Who the routes are from, asked of the table once for each. let mut sources: HashMap> = HashMap::default(); - let may_go = |prefix: &IpPrefix| { - session.families.carries(prefix) - && !self.own_announced.contains(prefix) - && limits.is_none_or(|limits| limits.allows(prefix)) - }; + let may_go = |prefix: &IpPrefix| prefix_may_go(session, &self.own_announced, prefix); for item in news { match item { // Dealt with by `follow`. @@ -1179,16 +1188,22 @@ impl Exporter { 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 { + for route in routes { + if !told_apart && !route.best { continue; } - let Some(source) = source_of(&self.rib, &mut sources, peer) else { + let Some(source) = source_of(&self.rib, &mut sources, route.peer) + else { continue; }; - if let Some(dressed) = learned.dressed(&attrs, source, version, &egress) + if let Some(dressed) = + learned.dressed(&route.attrs, source, version, &egress) { - wanted.push(((peer, path_id), attrs, dressed)); + wanted.push(Wanted { + key: (route.peer, route.path_id), + attrs: route.attrs, + dressed, + }); } } } @@ -1242,28 +1257,29 @@ impl Exporter { // again only if it changed; a new one gets the lowest // identifier not in use for the prefix. let mut has: Vec = 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 + for Wanted { + key, + attrs, + dressed, + } in wanted + { + let (sent, goes) = match had.iter().position(|sent| sent.key == key) { + Some(at) => had.swap_remove(at).renewed(&attrs), + None => { + let id = lowest_free_id(had.iter().chain(&has).map(|sent| sent.id)); + (Sent { key, id, attrs }, true) } - 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 }); + if goes { + announce( + dressed, + Nlri { + prefix, + path_id: sent.id, + }, + ); + } + has.push(sent); } withdrawn.extend(had.iter().map(|sent| Nlri { prefix, @@ -1274,61 +1290,42 @@ impl Exporter { routes.insert(prefix, has); } } - // A verdict is of the kind the session's bookkeeping is. - _ => {} + // A verdict is of the kind the session's bookkeeping is: + // `judge` asks the same `Learned` which. Were it not, the + // prefix would be left as the peer has it; better said than + // left in silence. + (verdict, announced) => { + debug_assert!(false, "a verdict is of the kind the bookkeeping is"); + warn!( + peer = %session.peer, + %prefix, + all_paths = matches!(verdict, Verdict::All(_)), + kept_all_paths = matches!(announced, Announced::All { .. }), + "A verdict on a prefix is of the wrong kind for the peer and is dropped" + ); + } } } // 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)); - } + within_limit(session, routes)?; + send_withdrawals(session, &towards, withdrawn).await?; // 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::(); - // The session's context for each IP version, with whether path - // identifiers go with the family's prefixes: made once a flush, - // not once an UPDATE. - let contexts = [IpVersion::Ipv4, IpVersion::Ipv6].map(|version| ParserContext { - four_octet_asn: towards.four_octet_asn, - ibgp: towards.ibgp, - add_path: if session.add_path_send.contains(AfiSafi::unicast(version)) { - vec![AfiSafi::unicast(version)] - } else { - Vec::new() - }, - }); + note_held(session, &self.advertised_metric, before); + let contexts = parser_contexts(session, &towards); 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::encoded_announcements( - &dressed.encoded, + let updates = dressed.encoded.announcements( dressed.next_hop, version, &routes, @@ -1338,9 +1335,8 @@ impl Exporter { session.send_encoded(update).await?; } announced += routes.len(); - note(before + announced); + note_held(session, &self.advertised_metric, before + announced); } - self.note_advertised(); Ok(()) } @@ -1386,9 +1382,7 @@ impl Exporter { 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)); - } + within_limit(session, routes)?; let towards = self.towards(); let Some(mut update) = build_announce(route, &towards) else { debug!( @@ -1434,6 +1428,88 @@ impl Exporter { } } +/// Whether a learned route for `prefix` may go to the peer at all, whatever +/// the route itself says: the session carries its family, we have no +/// route of our own for it (ours goes instead, as `own_announced` has +/// it), and it is within the export limits where there are any. Not a +/// method: `gather` holds the learned routes mutably while it asks this. +fn prefix_may_go( + session: &SessionHandle, + own_announced: &HashSet, + prefix: &IpPrefix, +) -> bool { + session.families.carries(prefix) + && !own_announced.contains(prefix) + && session + .export_limits + .as_deref() + .is_none_or(|limits| limits.allows(prefix)) +} + +/// The session's context for each IP version, with whether path +/// identifiers go with the family's prefixes: made once a flush, not once +/// an UPDATE. +fn parser_contexts(session: &SessionHandle, towards: &Towards) -> [ParserContext; 2] { + [IpVersion::Ipv4, IpVersion::Ipv6].map(|version| ParserContext { + four_octet_asn: towards.four_octet_asn, + ibgp: towards.ibgp, + add_path: if session.add_path_send.contains(AfiSafi::unicast(version)) { + vec![AfiSafi::unicast(version)] + } else { + Vec::new() + }, + }) +} + +/// Sends the peer the withdrawals of `withdrawn`, by IP version, and +/// counts them. +async fn send_withdrawals( + session: &SessionHandle, + towards: &Towards, + withdrawn: [Vec; 2], +) -> Result<(), Finished> { + 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?; + } + } + Ok(()) +} + +/// The export limit, if there is one, that the peer holding `routes` +/// routes from us would be over. +fn exceeded(session: &SessionHandle, routes: usize) -> Option { + session + .export_limits + .as_deref() + .and_then(|limits| limits.exceeded_by(routes)) +} + +/// Whether the peer may hold `routes` routes from us: if not, the session +/// is ended for it (`over_limit`), and nothing more is to go. +fn within_limit(session: &SessionHandle, routes: usize) -> Result<(), Finished> { + match exceeded(session, routes) { + Some(limit) => Err(over_limit(session, limit, routes)), + None => Ok(()), + } +} + +/// Notes that the peer holds `held` routes from us, for the status page +/// (the session's counters) and the metrics (`gauge`). +fn note_held(session: &SessionHandle, gauge: &IntGauge, held: usize) { + session.counters.set_advertised(held); + gauge.set(held as i64); +} + /// 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 { @@ -2000,6 +2076,17 @@ mod tests { use std::time::Duration; + /// A route the peer has not had gets the lowest identifier not in use + /// for its prefix, from one up: a gap left by a route that went is + /// filled before a new identifier is taken. + #[test] + fn a_new_route_gets_the_lowest_free_identifier() { + assert_eq!(lowest_free_id(std::iter::empty()), 1); + assert_eq!(lowest_free_id([1, 2].into_iter()), 3); + assert_eq!(lowest_free_id([1, 3].into_iter()), 2); + assert_eq!(lowest_free_id([3, 2].into_iter()), 1); + } + use bgp_rib::table::{PeerInfo, RibTable}; use tokio::sync::mpsc; @@ -2009,35 +2096,145 @@ mod tests { use crate::session::{Outbound, SessionGuard}; /// What a session was given to send, as a message: an encoded UPDATE - /// decoded as the peer decodes it, without ADD-PATH, in our own AS - /// (`ibgp`) or not: what attributes are expected and kept depends on + /// decoded as the peer decodes it, in our own AS (`ibgp`) or not, and + /// with path identifiers on its IPv4 routes (`told_apart`) or not: + /// what is expected and kept, and how the prefixes read, depend on /// which. - fn message_of(outbound: Outbound, ibgp: bool) -> Message { + fn decoded(outbound: Outbound, ibgp: bool, told_apart: bool) -> Message { match outbound { Outbound::Message(message) => message, Outbound::Encoded(bytes) => { let ctx = ParserContext { four_octet_asn: true, ibgp, - add_path: Vec::new(), + add_path: if told_apart { + vec![AfiSafi::IPV4_UNICAST] + } else { + Vec::new() + }, }; Message::decode(&ctx, &bytes).expect("an encoded UPDATE decodes") } } } - /// A registry and local RIB with an advertiser running, and a table to - /// take peer numbers from. + /// What a session was given to send, as a peer without ADD-PATH reads + /// it. + fn message_of(outbound: Outbound, ibgp: bool) -> Message { + decoded(outbound, ibgp, false) + } + + /// Folds what `update` says into `routes`, what the peer holds: what + /// it withdraws, in the body or in MP_UNREACH_NLRI, is gone, and what + /// it announces, in the body or in MP_REACH_NLRI, is there. + fn fold(update: &UpdateMessage, routes: &mut HashSet) { + for nlri in &update.withdrawn_routes { + routes.remove(&nlri.prefix); + } + routes.extend(update.announced_routes.iter().map(|nlri| nlri.prefix)); + for attribute in &update.path_attributes { + match attribute { + PathAttribute::MpReachNlri(reach) => { + routes.extend(reach.prefixes.iter().map(|nlri| nlri.prefix)); + } + PathAttribute::MpUnreachNlri(unreach) => { + for nlri in &unreach.prefixes { + routes.remove(&nlri.prefix); + } + } + _ => {} + } + } + } + + /// What was agreed with a peer in AS `peer_asn`, in our own AS or not: + /// both IP versions, four-octet AS numbers, and nothing more. + fn negotiated(peer_asn: u32, ibgp: bool) -> Negotiated { + Negotiated { + peer_asn, + ibgp, + 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, + } + } + + /// A handle to a session with the peer at 192.0.2.`n`, as `negotiated`, + /// whose routes in the table are `rib_peer`'s and whose queue is `tx`. + fn handle( + n: u8, + negotiated: &Negotiated, + rib_peer: PeerIndex, + tx: mpsc::Sender, + ) -> SessionHandle { + SessionHandle::new( + PeerId { + addr: (Ipv4Addr::new(192, 0, 2, n), 179).into(), + asn: negotiated.peer_asn, + }, + None, + Direction::Inbound, + (Ipv4Addr::new(192, 0, 2, 100), 179).into(), + negotiated, + rib_peer, + tx, + ) + } + + /// A peer of another AS at 192.0.2.`n`, as the table knows it. + fn peer_info(n: u8, asn: u32) -> PeerInfo { + 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, + } + } + + /// An export policy that tells the peer of learned routes: the best + /// of each prefix, or every one (`all_paths`). + fn learned(all_paths: bool) -> Arc { + Arc::new(ExportPolicy { + learned_routes: true, + all_paths, + ..ExportPolicy::default() + }) + } + + /// A route from AS `asn`, as the table holds its attributes. + fn route_from(rib: &RibTable, asn: u32) -> Arc { + let came = [ + PathAttribute::Origin(OriginPathAttribute::IGP), + AsPathAttribute::from_asns(vec![asn]), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))), + ]; + rib.attrs().intern(&came, None).unwrap() + } + + /// A registry and local RIB with an advertiser running over a table. struct Fixture { registry: SessionRegistry, local_rib: LocalRib, - table: RibTable, + /// The table the advertiser follows. + rib: Arc, } /// The far end of a session's queue: what the session's task would /// write to the peer. struct Queue { + peer: PeerId, outbound: mpsc::Receiver, + /// Whether IPv4 routes go to the peer with path identifiers. + told_apart: bool, /// What the peer holds, going by what it has been sent. routes: HashSet, _registration: SessionGuard, @@ -2047,9 +2244,10 @@ mod tests { fn new() -> Self { let registry = SessionRegistry::new(); let local_rib = LocalRib::spawn(); + let rib = Arc::new(RibTable::new(4)); spawn_advertiser( local_rib.clone(), - Arc::new(RibTable::new(4)), + Arc::clone(&rib), registry.clone(), 65000, None, @@ -2059,54 +2257,36 @@ mod tests { Self { registry, local_rib, - table: RibTable::new(1), + rib, } } - /// A session with peer number `n`, whose queue holds `capacity` - /// messages. + /// A session with peer number `n`, told of our own routes only, + /// 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, - }; + self.session_with(n, capacity, |_| {}) + } + + /// A session with peer number `n`, whose queue holds `capacity` + /// messages, with the handle as `set_up` leaves it. + fn session_with( + &self, + n: u8, + capacity: usize, + set_up: impl FnOnce(&mut SessionHandle), + ) -> Queue { + let negotiated = negotiated(65000 + u32::from(n), false); 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, - }) + .rib + .add_peer(peer_info(n, negotiated.peer_asn)) .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, - ); + let mut handle = handle(n, &negotiated, rib_peer, tx); + set_up(&mut handle); Queue { + peer: handle.peer, outbound, + told_apart: handle.add_path_send.contains(AfiSafi::IPV4_UNICAST), routes: HashSet::default(), _registration: self.registry.register(handle), } @@ -2129,22 +2309,25 @@ mod tests { } impl Queue { + /// The next UPDATE queued for the peer, taken in as the peer would. + async fn next_update(&mut self) -> UpdateMessage { + let next = tokio::time::timeout(Duration::from_secs(10), self.outbound.recv()) + .await + .expect("the peer is sent something") + .expect("the session's queue is open"); + let Message::Update(update) = decoded(next, false, self.told_apart) else { + panic!("expected an UPDATE"); + }; + fold(&update, &mut self.routes); + update + } + /// 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) { let reading = async { while self.routes != *expected { - let Some(Message::Update(update)) = - self.outbound.recv().await.map(|m| message_of(m, false)) - 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); - } + self.next_update().await; } }; if tokio::time::timeout(Duration::from_secs(10), reading) @@ -2159,6 +2342,18 @@ mod tests { ); } } + + /// Asserts that the peer is sent nothing more for a while. + async fn quiet(&mut self) { + if let Ok(more) = + tokio::time::timeout(Duration::from_millis(300), self.outbound.recv()).await + { + panic!( + "the peer was sent more: {:?}", + more.map(|m| decoded(m, false, self.told_apart)) + ); + } + } } /// A route from a router of our AS reaches a route reflector client @@ -2198,39 +2393,10 @@ mod tests { 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() - }); + let mut handle = handle(2, &negotiated(65000, true), rib_peer, tx); + handle.export = learned(false); handle.reflector_client = true; let _registration = registry.register(handle); @@ -2278,14 +2444,7 @@ mod tests { Ipv4Addr::new(10, 0, 0, 100), watch::channel(true).1, ); - let peer = |n: u8| PeerInfo { - asn: 64496 + u32::from(n), - router_id: Ipv4Addr::new(10, 0, 0, n), - address: IpAddr::V4(Ipv4Addr::new(192, 0, 2, n)), - interface: 0, - ibgp: false, - reflector_client: false, - }; + let peer = |n: u8| peer_info(n, 64496 + u32::from(n)); // A peer with 300 routes that share three sets of attributes, // spread over the table's shards. let from = rib.add_peer(peer(1)).unwrap(); @@ -2303,39 +2462,10 @@ mod tests { } // A new session that is told of learned routes. - let negotiated = Negotiated { - peer_asn: 64498, - 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)).unwrap(); let (tx, mut outbound) = mpsc::channel(1024); - let mut handle = SessionHandle::new( - PeerId { - addr: (Ipv4Addr::new(192, 0, 2, 2), 179).into(), - asn: 64498, - }, - 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() - }); + let mut handle = handle(2, &negotiated(64498, false), rib_peer, tx); + handle.export = learned(false); let _registration = registry.register(handle); let mut updates = 0; @@ -2345,7 +2475,7 @@ mod tests { match outbound.recv().await.map(|m| message_of(m, false)) { Some(Message::Update(update)) if !update.announced_routes.is_empty() => { updates += 1; - routes.extend(update.announced_routes.iter().map(|nlri| nlri.prefix)); + fold(&update, &mut routes); } Some(_) => {} None => panic!("the session's queue closed"), @@ -2357,6 +2487,121 @@ mod tests { assert_eq!(updates, 3, "one UPDATE for each set of attributes"); } + /// On a session that is told of every route, the routes of a prefix + /// are told apart by path identifiers of our giving (RFC 7911): a + /// route the peer has not had gets the lowest one not in use for the + /// prefix, and keeps it for as long as it is announced. A route that + /// goes is withdrawn under its identifier, and the others are not + /// sent again for it coming or going. + #[tokio::test] + async fn two_routes_for_a_prefix_go_with_their_own_identifiers_and_keep_them() { + let fixture = Fixture::new(); + let prefix = nth_prefix(1); + let first = fixture.rib.add_peer(peer_info(1, 64497)).unwrap(); + let second = fixture.rib.add_peer(peer_info(2, 64498)).unwrap(); + fixture + .rib + .announce(first, &route_from(&fixture.rib, 64497), &[prefix]); + let mut peer = fixture.session_with(3, 16, |handle| { + handle.export = learned(true); + handle.add_path_send = FamilySet::from_iter([AfiSafi::IPV4_UNICAST]); + }); + let as_path = |update: &UpdateMessage| { + update + .path_attributes + .iter() + .find(|attribute| matches!(attribute, PathAttribute::AsPath(_))) + .cloned() + .expect("an AS path") + }; + + // The one route of the prefix, as the table is given out. + let update = peer.next_update().await; + assert_eq!(update.announced_routes, [Nlri { prefix, path_id: 1 }]); + assert_eq!( + as_path(&update), + AsPathAttribute::from_asns(vec![65000, 64497]) + ); + + // A second route for the prefix: the next identifier, and the + // first route is not sent again. + fixture + .rib + .announce(second, &route_from(&fixture.rib, 64498), &[prefix]); + let update = peer.next_update().await; + assert_eq!(update.announced_routes, [Nlri { prefix, path_id: 2 }]); + assert_eq!( + as_path(&update), + AsPathAttribute::from_asns(vec![65000, 64498]) + ); + + // The first route goes: withdrawn under its identifier, with the + // other left as it is. + fixture.rib.withdraw(first, &[prefix]); + let update = peer.next_update().await; + assert_eq!(update.withdrawn_routes, [Nlri { prefix, path_id: 1 }]); + assert!(update.announced_routes.is_empty(), "{update:?}"); + peer.quiet().await; + } + + /// Once the table is given out to a peer anew, everything it held is + /// in doubt until it is heard of again, and what is not is withdrawn + /// after the table: here a route that went from the table while the + /// peer was being sent our own routes again, before the table. + #[tokio::test] + async fn what_is_not_heard_of_again_after_a_reset_is_withdrawn() { + let fixture = Fixture::new(); + let from = fixture.rib.add_peer(peer_info(1, 64496)).unwrap(); + let learned_routes = [nth_prefix(1), nth_prefix(2), nth_prefix(3)]; + fixture + .rib + .announce(from, &route_from(&fixture.rib, 64496), &learned_routes); + // Two routes of our own, to a peer whose queue holds one message: + // the exporter is held at the second until the peer reads the + // first, and does not look at the table anew before that. + fixture.announce(100).await; + fixture.announce(101).await; + let mut peer = fixture.session_with(2, 1, |handle| handle.export = learned(false)); + let mut all = fixture.routes().await; + all.extend(learned_routes); + peer.read_until_it_has(&all).await; + + // The peer asks for everything again. Our own routes go first; + // once the peer has read the first, the exporter is held at the + // second, and a route that goes from the table now is one the + // table, when the exporter comes to it, no longer has. + fixture.registry.request_refresh(&peer.peer); + loop { + let update = peer.next_update().await; + if update + .announced_routes + .iter() + .any(|nlri| nlri.prefix == nth_prefix(100)) + { + break; + } + } + fixture.rib.withdraw(from, &[nth_prefix(2)]); + + // The table comes, and the withdrawal after it. + let mut again = HashSet::default(); + loop { + let update = peer.next_update().await; + again.extend(update.announced_routes.iter().map(|nlri| nlri.prefix)); + if !update.withdrawn_routes.is_empty() { + assert_eq!(update.withdrawn_routes, [Nlri::from(nth_prefix(2))]); + break; + } + } + assert!( + again.contains(&nth_prefix(1)) && again.contains(&nth_prefix(3)), + "the table was to come before the withdrawal, not {again:?}" + ); + all.remove(&nth_prefix(2)); + assert_eq!(peer.routes, all); + peer.quiet().await; + } + /// 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. @@ -2446,60 +2691,16 @@ mod tests { 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 = prefixes.iter().copied().map(Nlri::from).collect(); - rib.announce(customer, &attrs, &nlri); + let customer = rib.add_peer(peer_info(1, 64496)).unwrap(); + rib.announce(customer, &route_from(&rib, 64496), prefixes); - 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 rib_peer = rib.add_peer(peer_info(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, - ); + let mut handle = handle(2, &negotiated(64497, false), rib_peer, tx); handle.role = Some(PeerRole::Peer); - handle.export = Arc::new(ExportPolicy { - learned_routes: true, - all_paths: false, - ..ExportPolicy::default() - }); + handle.export = learned(false); handle.export_limits = Some(Arc::new(limits)); let ending = handle.listen_for_end(); LimitedPeer { @@ -2525,17 +2726,7 @@ mod tests { tokio::time::timeout(Duration::from_millis(500), outbound.recv()).await { if let Message::Update(update) = message_of(message, false) { - 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)); - } + fold(&update, &mut routes); } } routes diff --git a/crates/server/src/export.rs b/crates/server/src/export.rs index 1bccdf4..2c2a101 100644 --- a/crates/server/src/export.rs +++ b/crates/server/src/export.rs @@ -33,9 +33,9 @@ use std::net::{IpAddr, Ipv4Addr}; use bgp_packet::message::{ - AsPathAttribute, AsPathSegment, BGP_MARKER, LocalPrefPathAttribute, MpReachNlriPathAttribute, + AsPathAttribute, AsPathSegment, LocalPrefPathAttribute, MpReachNlriPathAttribute, 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::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 /// does and as what. #[derive(Debug, Clone, Copy)] -pub struct Egress<'a> { +pub(crate) struct Egress<'a> { pub towards: Towards, pub peer_asn: u32, /// 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. #[derive(Debug, Clone, Copy)] -pub struct Reflection { +pub(crate) struct Reflection { /// The cluster this router reflects routes for. pub cluster_id: Ipv4Addr, /// Whether the peer is a route reflector client of ours. @@ -81,7 +81,7 @@ pub struct Reflection { /// Where a learned route came from. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct Source { +pub(crate) struct Source { pub peer: PeerIndex, /// Whether that peer is in our own AS. pub ibgp: bool, @@ -95,7 +95,7 @@ pub struct Source { /// A route ready to be announced on a session: its attributes, but for /// the ones that depend on which prefixes go with them. #[derive(Debug, Clone, PartialEq, Eq)] -pub struct Outgoing { +pub(crate) struct Outgoing { pub attributes: Vec, pub next_hop: NlriNextHop, } @@ -396,6 +396,14 @@ pub(crate) fn finish(attributes: &mut Vec, towards: &Towards) { /// lengths every UPDATE has. 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 /// length, and the address. 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 /// type code. #[derive(Debug)] -pub struct EncodedAttributes { - pub bytes: Bytes, - pub at_next_hop: usize, - pub at_mp_reach: usize, +pub(crate) struct EncodedAttributes { + bytes: Bytes, + at_next_hop: usize, + at_mp_reach: usize, } impl EncodedAttributes { /// Encodes `attributes`, which are `finish`ed, for a session with /// `four_octet_asn`. `None` if they cannot be encoded. - pub fn encode(attributes: &[PathAttribute], four_octet_asn: bool) -> Option { + pub(crate) fn encode(attributes: &[PathAttribute], four_octet_asn: bool) -> Option { let ctx = ParserContext { four_octet_asn, ibgp: false, @@ -444,9 +452,81 @@ impl EncodedAttributes { }) } - pub fn octets(&self) -> usize { + pub(crate) fn octets(&self) -> usize { 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 { + // 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, @@ -458,84 +538,12 @@ const NEXT_HOP_ROOM: usize = 64; /// so it runs over by at most one. 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 { - 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 /// `outgoing`: as few as they fit in. `attributes_octets` is how long /// `outgoing`'s attributes are on this session's wire. /// -/// What [`encoded_announcements`] does, built as messages: kept as what -/// its bytes are checked against. +/// What [`EncodedAttributes::announcements`] does, built as messages: +/// kept as what its bytes are checked against. #[cfg(test)] pub(crate) fn announcements( outgoing: &Outgoing, @@ -544,9 +552,6 @@ pub(crate) fn announcements( towards: &Towards, attributes_octets: usize, ) -> Vec { - // 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 at_once = (room / nlri_octets(version)).max(1); let in_body = matches!(outgoing.next_hop, NlriNextHop::Ipv4(_)) && version == IpVersion::Ipv4; @@ -684,31 +689,47 @@ mod tests { PathAttribute::Communities(CommunitiesPathAttribute(communities.to_vec())) } - /// What `encoded_announcements` writes is, byte for byte, what the - /// messages `announcements` builds encode to: with the next hop in - /// its own attribute (IPv4) and in MP_REACH_NLRI (IPv6), with and - /// without path identifiers, to two-octet and four-octet AS speakers - /// (so with the attributes around NEXT_HOP, MP_REACH_NLRI and - /// AS4_PATH in order), and over as many messages when the routes do - /// not fit in one. + /// What `EncodedAttributes::announcements` writes is, byte for byte, + /// what the messages `announcements` builds encode to: with the next + /// hop in its own attribute (IPv4) and in MP_REACH_NLRI (IPv6, and + /// IPv4 with an IPv6 next hop, RFC 8950), with and without path + /// identifiers, to two-octet and four-octet AS speakers (so with the + /// attributes around NEXT_HOP, MP_REACH_NLRI and AS4_PATH in order), + /// and over as many messages when the routes do not fit in one. #[test] fn encoded_announcements_are_the_messages_on_the_wire() { use std::net::Ipv6Addr; + // The routes' IP version, their next hop, and whether the session + // takes IPv4 routes with IPv6 next hops. + let layouts = [ + ( + IpVersion::Ipv4, + NlriNextHop::Ipv4(Ipv4Addr::new(192, 0, 2, 1)), + false, + ), + ( + IpVersion::Ipv6, + NlriNextHop::Ipv6WithLl { + global: Ipv6Addr::new(0x2001, 0xdb8, 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 in [IpVersion::Ipv4, IpVersion::Ipv6] { + for (version, next_hop, extended_next_hop) in layouts { let towards = Towards { four_octet_asn, + extended_next_hop, ..towards(None, false) }; - let next_hop = match version { - IpVersion::Ipv4 => NlriNextHop::Ipv4(Ipv4Addr::new(192, 0, 2, 1)), - IpVersion::Ipv6 => NlriNextHop::Ipv6WithLl { - global: Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1), - link_local: Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 1), - }, - }; let attributes = vec![ PathAttribute::Origin(OriginPathAttribute::IGP), AsPathAttribute::from_asns(vec![4_200_000_000, 3356, 64700]), @@ -753,7 +774,7 @@ mod tests { let mut finished = attributes.clone(); finish(&mut finished, &towards); 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 { attributes, @@ -771,7 +792,8 @@ mod tests { .collect(); 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"); for update in &expected { diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index 114ffe0..d7ac442 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -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 /// address. pub fn metric_label(&self) -> String { @@ -305,6 +303,8 @@ impl SessionHandle { .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 { self.outbound.same_channel(&other.outbound) } @@ -460,10 +460,6 @@ impl SessionRegistry { 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` /// says: for good, or to restart (see [`Leave`]). Waits up to `grace` /// for the sessions to finish; returns how many were still open.