From 3b22e318a229f4392cc8e4dad88d8d1b230f4328 Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Tue, 6 Oct 2026 11:15:58 +0200 Subject: [PATCH] Server: one way to work on the table off-thread, and smaller tidying Removing a peer's routes, sweeping its stale ones and sweeping the attribute sets were each written out as their own spawn_blocking, and each dealt with a failed task differently: the periodic sweep dropped a panic on the floor, and a restarting peer whose kept routes could not be removed was logged as "removed 0 routes". off_thread does it once, and every caller logs a failure. Also: - the hard-reset wrapping of a NOTIFICATION is one function in the FSM; - the two ways to close every session are one, and the registry has one leave() rather than leave() and shutdown(); - the dialer says why a session ended with SessionEnd::reason, and no longer turns a result into a result; - as_path_prepend is parsed in one place, and the limits in error texts come from the constants rather than being written out; - router_id and cluster_id are parsed by one function; - the exporter builds its Egress in one place; - the fields a session keeps only to hand to its handle are fewer: route_refresh is a bool, not a copy of the families; - counting the stale routes removed cannot underflow; - docs that had gone stale: the proto still said there was no route reflection, the advertiser's module doc said it announced only our own routes, from_server listed two of its five sources, and so on. Tests for what had none: which session ends count as restarts, the Hard Reset wrapping, the graceful restart settings, the dialer's jitter, withdrawals in MP_UNREACH_NLRI, and AS_TRANS on a two-octet session. Co-Authored-By: Claude Fable 5.1 --- crates/integration-tests/tests/announce.rs | 5 +- crates/server/proto/config.proto | 13 +- crates/server/src/advertise.rs | 62 +++++----- crates/server/src/collision.rs | 9 +- crates/server/src/config.rs | 46 ++++--- crates/server/src/connection.rs | 54 +++++---- crates/server/src/dialer.rs | 27 ++++- crates/server/src/error.rs | 10 +- crates/server/src/export.rs | 30 ++++- crates/server/src/fsm.rs | 27 +++-- crates/server/src/policy.rs | 21 ++-- crates/server/src/restart.rs | 132 ++++++++++++++++++--- crates/server/src/server.rs | 24 ++-- crates/server/src/session.rs | 12 +- 14 files changed, 335 insertions(+), 137 deletions(-) diff --git a/crates/integration-tests/tests/announce.rs b/crates/integration-tests/tests/announce.rs index 5dcf10f..1f6163f 100644 --- a/crates/integration-tests/tests/announce.rs +++ b/crates/integration-tests/tests/announce.rs @@ -393,7 +393,10 @@ async fn test_shutdown_sends_cease_to_every_peer() -> Result<()> { let mut b = established_peer_with_id(&server, Ipv4Addr::new(10, 0, 0, 2)).await?; wait_for_sessions(&server, 2).await?; - let left = server.registry.shutdown(Duration::from_secs(2)).await; + let left = server + .registry + .leave(bgp_server::restart::Leave::Stop, Duration::from_secs(2)) + .await; assert_eq!(left, 0, "sessions still open after shutdown"); assert!(server.registry.is_shutting_down()); diff --git a/crates/server/proto/config.proto b/crates/server/proto/config.proto index e107c9e..0c09164 100644 --- a/crates/server/proto/config.proto +++ b/crates/server/proto/config.proto @@ -233,7 +233,8 @@ message Rpki { } // A route originated by this router. It is announced with an AS_PATH of the -// local AS and, on each session, that session's local address as next hop. +// local AS (empty, to a peer in our own AS) and, on each session, that +// session's local address as next hop. message OriginatedRoute { // The prefix in CIDR notation, e.g. "192.0.2.0/24" or "2001:db8::/32". string prefix = 1; @@ -376,8 +377,9 @@ message Peer { // // - Never back to the peer a route came from, nor to a peer whose AS is // on its path already. -// - A route from a peer in our own AS is not passed to another such peer -// (there is no route reflection). +// - A route from a peer in our own AS is not passed to another such peer, +// unless one of them is a route reflector client of ours +// (Peer.route_reflector_client). // - Where peers have roles (RFC 9234): a route learned from a provider, a // peer or a route server goes on to customers only. So with roles on // all peers, turning this on makes us transit for our customers and for @@ -453,7 +455,7 @@ message ImportPolicy { // one, as from an upstream that is asked to send it. bool accept_default_route = 4; - // The longest prefixes taken. Default to 24 and 48, the longest that are + // The longest prefixes taken. Defaults to 24 and 48, the longest that are // routed between networks. uint32 max_ipv4_prefix_length = 5; uint32 max_ipv6_prefix_length = 6; @@ -565,7 +567,8 @@ enum AddPath { ADD_PATH_OFF = 0; // Accept several routes for a prefix from the peer. ADD_PATH_RECEIVE = 1; - // Send the peer several routes for a prefix. + // Offer to send the peer several routes for a prefix. As yet only the + // best is sent all the same. ADD_PATH_SEND = 2; ADD_PATH_BOTH = 3; } diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 1d31263..83e5230 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -1,5 +1,6 @@ -//! Announces the routes this router originates to every established -//! session. +//! Announces routes to every established session: those this router +//! originates, and, to the sessions whose export policy says so, the best +//! of those learned from others. //! //! Each session has an exporter of its own: a task that sends the session //! the originated routes and then every change to them. An exporter waits @@ -16,7 +17,7 @@ //! and stops it when the session goes. //! //! To a peer above or beside us, the export limits are a check on all of -//! this ([`ExportLimits`]): routes for prefixes outside them are not +//! this ([`crate::policy::ExportLimits`]): routes for prefixes outside them are not //! announced, whatever else says they are to be, and if more routes were //! to be announced than they allow, none of them is and the session ends. //! @@ -127,7 +128,7 @@ impl Advertiser { } Ok(SessionEvent::RefreshRequested(session)) => { if let Some(exporter) = self.exporters.get(&session.peer) { - debug!(peer = %session.peer, "Route refresh requested, re-sending local routes"); + debug!(peer = %session.peer, "Route refresh requested, re-sending our routes"); exporter.resend.notify_one(); } } @@ -390,7 +391,7 @@ impl Exporter { /// Runs until the session closes or the local RIB stops; the /// advertiser stops it when it hears the session has gone. async fn run(mut self) { - info!(peer = %self.session.peer, "Session up, sending local routes"); + info!(peer = %self.session.peer, "Session up, sending it our routes"); let _ = self.export().await; } @@ -460,6 +461,29 @@ impl Exporter { Ok(()) } + /// What every route on its way to `session` goes through. Not a method: + /// `follow` holds the learned routes mutably while it uses this. + fn egress( + session: &SessionHandle, + control: Option, + cluster_id: Ipv4Addr, + towards: Towards, + ) -> Egress<'_> { + Egress { + towards, + peer_asn: session.peer.asn, + rib_peer: session.rib_peer, + policy: &session.export, + groups: &session.groups, + control, + reflection: Reflection { + cluster_id, + client: session.reflector_client, + peer_router_id: session.peer_router_id, + }, + } + } + /// The session as far as it decides what goes out on it and how. fn towards(&self) -> Towards { let session = &self.session; @@ -510,19 +534,7 @@ impl Exporter { let session = &self.session; let limits = session.export_limits.as_deref(); - let egress = Egress { - towards, - peer_asn: session.peer.asn, - rib_peer: session.rib_peer, - policy: &session.export, - groups: &session.groups, - control: self.control, - reflection: Reflection { - cluster_id: self.cluster_id, - client: session.reflector_client, - peer_router_id: session.peer_router_id, - }, - }; + let egress = Self::egress(session, self.control, self.cluster_id, towards); // What the peer is to end up with for each prefix heard of: the // last word on it counts. While the whole table is being given out // anew, which is shard by shard, this is kept until it is all @@ -720,19 +732,7 @@ impl Exporter { }; // What every route on its way to this peer goes through: whether // its communities let it go there, and what it is marked with. - let egress = Egress { - towards, - peer_asn: session.peer.asn, - rib_peer: session.rib_peer, - policy: &session.export, - groups: &session.groups, - control: self.control, - reflection: Reflection { - cluster_id: self.cluster_id, - client: session.reflector_client, - peer_router_id: session.peer_router_id, - }, - }; + let egress = Self::egress(session, self.control, self.cluster_id, towards); if !egress.dress(&mut update.path_attributes) { debug!( peer = %session.peer, diff --git a/crates/server/src/collision.rs b/crates/server/src/collision.rs index cecb29c..78e6a02 100644 --- a/crates/server/src/collision.rs +++ b/crates/server/src/collision.rs @@ -105,10 +105,10 @@ impl CollisionTable { /// Call when a connection receives the peer's OPEN. `one_identifier` /// says that the address is that of one router, which holds for a - /// configured peer. `None` means this - /// connection lost and must be closed with Cease / Connection Collision - /// Resolution. Otherwise any rival has been told to close and this - /// connection holds the slot until the guard drops. + /// configured peer. `None` means this connection lost and must be + /// closed with Cease / Connection Collision Resolution. Otherwise any + /// rival has been told to close and this connection holds the slot + /// until the guard drops. pub fn admit( &self, local_id: Ipv4Addr, @@ -143,6 +143,7 @@ impl CollisionTable { } if let Some(kill) = existing.kill.take() { COLLISIONS.with_label_values(&["existing"]).inc(); + // Fails only if the rival has gone already. let _ = kill.send(()); } } diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index c4d49f3..35e36ff 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -119,10 +119,7 @@ impl Config { } pub fn from_proto(proto: BgpConfig) -> Result { - // Peers refuse an identifier of 0.0.0.0, as we do theirs. - let router_id = Ipv4Addr::from_str(&proto.router_id) - .ok() - .filter(|id| !id.is_unspecified()) + let router_id = parse_identifier(&proto.router_id) .ok_or_else(|| ConfigError::RouterId(proto.router_id.clone()))?; if proto.local_asn == 0 { @@ -131,9 +128,7 @@ impl Config { let cluster_id = if proto.cluster_id.is_empty() { router_id } else { - Ipv4Addr::from_str(&proto.cluster_id) - .ok() - .filter(|id| !id.is_unspecified()) + parse_identifier(&proto.cluster_id) .ok_or_else(|| ConfigError::ClusterId(proto.cluster_id.clone()))? }; @@ -302,6 +297,14 @@ impl Config { } } +/// A BGP identifier (RFC 4271 §4.2) as the configuration writes one: a +/// dotted quad other than 0.0.0.0, which peers refuse, as we do theirs. +fn parse_identifier(text: &str) -> Option { + Ipv4Addr::from_str(text) + .ok() + .filter(|id| !id.is_unspecified()) +} + fn parse_kernel(kernel: &proto::Kernel) -> FibConfig { let defaults = FibConfig::default(); FibConfig { @@ -318,6 +321,14 @@ fn parse_kernel(kernel: &proto::Kernel) -> FibConfig { } } +/// How many more times our AS goes on a path, if `prepend` is a number of +/// times we allow. +fn parse_prepend(prepend: u32) -> Option { + u8::try_from(prepend) + .ok() + .filter(|prepend| u32::from(*prepend) <= MAX_AS_PATH_PREPEND) +} + fn parse_export(export: &proto::ExportPolicy) -> Result { Ok(ExportPolicy { learned_routes: export.learned_routes, @@ -326,10 +337,10 @@ fn parse_export(export: &proto::ExportPolicy) -> Result Result OriginPathAttribute::INCOMPLETE, _ => return Err(OriginateProblem::Origin(route.origin.clone())), }; - let prepend = u8::try_from(route.as_path_prepend) - .ok() - .filter(|&n| u32::from(n) <= MAX_AS_PATH_PREPEND) - .ok_or(OriginateProblem::Prepend { - max: MAX_AS_PATH_PREPEND, - })?; + let prepend = parse_prepend(route.as_path_prepend).ok_or(OriginateProblem::Prepend { + value: route.as_path_prepend, + max: MAX_AS_PATH_PREPEND, + })?; let communities = route .communities .iter() @@ -893,7 +902,7 @@ mod tests { ); assert_eq!( problem("prefix: \"192.0.2.0/24\" as_path_prepend: 17"), - OriginateProblem::Prepend { max: 16 } + OriginateProblem::Prepend { value: 17, max: 16 } ); assert!(matches!( problem("prefix: \"192.0.2.0/24\" communities: \"70000:1\""), @@ -1249,7 +1258,6 @@ mod tests { ConfigError::RpkiCachePort("192.0.2.10".into()) ); - // A peer can be exempt. // A peer's paths are verified only if it is said what the peer is. let role_of = |body: &str| { parse(&format!( diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index 923a3f1..eb53589 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -285,6 +285,7 @@ pub async fn run_session( .. } = established; // The peer has no session to count its routes for. + // Fails only if the session never set it, which is nothing to do. let _ = PEER_ROUTES.remove_label_values(&[session.refusals.label()]); if leaving == Some(Leave::Restart) { // The daemon is about to go, and its routes with it; the @@ -352,7 +353,7 @@ struct Established { held: Held, /// The address families the peer can be asked to send again, if it can /// be asked at all. - refreshable: Option, + route_refresh: bool, /// The address families of the session. families: FamilySet, /// Graceful restart (RFC 4724), if both sides offered it. @@ -446,10 +447,8 @@ impl PeerRoutes { /// Removes the routes and says what that did. async fn remove(mut self) -> Outcome { - let (table, peer) = (Arc::clone(&self.table), self.peer); - // Looks at every prefix, which for a full table is a tenth of a - // second: too long to hold up the other sessions on this thread. - match tokio::task::spawn_blocking(move || Self::remove_now(&table, peer)).await { + let peer = self.peer; + match off_thread(&self.table, move |table| table.remove_peer(peer)).await { Ok(outcome) => { self.removed = true; outcome @@ -476,7 +475,7 @@ impl PeerRoutes { .filter(|version| !versions.contains(version)) .collect(); let removed = table.sweep_stale(peer, &others).removed; - marked - removed + marked.saturating_sub(removed) }) .await; match kept { @@ -499,12 +498,31 @@ impl PeerRoutes { fn remove_now(table: &RibTable, peer: PeerIndex) -> Outcome { let outcome = table.remove_peer(peer); - // Most of the peer's attribute sets were its alone. table.attrs().sweep(); outcome } } +/// Runs `work` on the table on a thread that can take its time, and then +/// sweeps the attribute sets that nothing refers to any more. For what +/// looks at every prefix: removing a peer's routes, which for a full +/// table takes a tenth of a second, too long to hold up the sessions that +/// share this thread. Fails only if `work` panicked. +pub(crate) async fn off_thread( + table: &Arc, + work: impl FnOnce(&RibTable) -> T + Send + 'static, +) -> Result { + let table = Arc::clone(table); + tokio::task::spawn_blocking(move || { + let outcome = work(&table); + // Most of the attribute sets the work let go of were its routes' + // alone. + table.attrs().sweep(); + outcome + }) + .await +} + impl Drop for PeerRoutes { fn drop(&mut self) { if self.removed { @@ -1017,9 +1035,7 @@ impl Session { since: Instant::now(), validation, held: Held::default(), - refreshable: negotiated - .route_refresh - .then(|| negotiated.families.clone()), + route_refresh: negotiated.route_refresh, families: negotiated.families.clone(), graceful_restart: negotiated.graceful_restart.clone(), end_of_rib: FamilySet::default(), @@ -1218,11 +1234,10 @@ impl Session { } // Invalid routes that there was no room to hold may be acceptable // now, and only the peer has them. - if established.held.take_overflowed() - && let Some(families) = &established.refreshable - { + if established.held.take_overflowed() && established.route_refresh { info!(peer = %established.peer_id, "Asking the peer to send its routes again"); - return families + return established + .families .iter() .map(|family| { Action::Send(Message::RouteRefresh(RouteRefreshMessage::request(family))) @@ -1285,12 +1300,7 @@ fn sweep( ) { let (rib, done) = (Arc::clone(rib), done.clone()); tokio::spawn(async move { - let swept = tokio::task::spawn_blocking(move || { - let outcome = rib.sweep_stale(peer, &versions); - rib.attrs().sweep(); - outcome - }) - .await; + let swept = off_thread(&rib, move |table| table.sweep_stale(peer, &versions)).await; match swept { // The session may have ended meanwhile. Ok(outcome) => { @@ -1302,7 +1312,9 @@ fn sweep( } /// What the rest of the server has for an established session: a message -/// to send to the peer, or word that the validation data has changed. +/// to send to the peer, a request to end the session, word that the +/// validation data has changed, the outcome of a sweep of stale routes, +/// or the time being up for the routes kept from the peer's last session. async fn from_server(established: &mut Option) -> Wake { let Some(Established { outbound, diff --git a/crates/server/src/dialer.rs b/crates/server/src/dialer.rs index e6a181c..a73211b 100644 --- a/crates/server/src/dialer.rs +++ b/crates/server/src/dialer.rs @@ -121,7 +121,11 @@ async fn dial_loop(peer: PeerConfig, server: Arc, connect_retry: if end.established { backoff = connect_retry; } - end.result.is_ok().then_some(()).ok_or_else(|| end.reason()) + if end.result.is_ok() { + Ok(()) + } else { + Err(end.reason()) + } } Err(e) => Err(format!("connection lost at once: {e}")), }, @@ -150,3 +154,24 @@ async fn dial_loop(peer: PeerConfig, server: Arc, connect_retry: backoff = (backoff * 2).min(MAX_BACKOFF); } } + +#[cfg(test)] +mod tests { + use super::*; + + /// The wait before dialling again is shortened by up to a quarter, by + /// chance, so that peers that went away together are not all dialled + /// at the same moment (RFC 4271 §10). + #[test] + fn the_wait_is_jittered_by_up_to_a_quarter() { + let wait = Duration::from_secs(100); + let mut shortest = wait; + for _ in 0..200 { + let jittered = jittered(wait); + assert!(jittered <= wait, "{jittered:?}"); + assert!(jittered >= wait.mul_f64(0.75), "{jittered:?}"); + shortest = shortest.min(jittered); + } + assert!(shortest < wait, "the wait varies"); + } +} diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index f7e9c7d..d00d27d 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -284,7 +284,7 @@ pub enum PeerProblem { /// What is wrong with a `graceful_restart`. #[derive(Debug, Clone, PartialEq, Eq, Error)] pub enum GracefulRestartProblem { - #[error("restart_time_seconds is {0}; it is at most 4095")] + #[error("restart_time_seconds is {0}; it is at most {max}", max = crate::restart::MAX_RESTART_TIME)] RestartTime(u32), } @@ -327,8 +327,8 @@ pub enum ImportProblem { pub enum ExportProblem { #[error(transparent)] Marks(#[from] crate::marks::MarksProblem), - #[error("as_path_prepend of {0} is more than 16")] - Prepend(u32), + #[error("as_path_prepend of {value} is more than {max}")] + Prepend { value: u32, max: u32 }, } /// What is wrong with one `originate` entry. @@ -338,8 +338,8 @@ pub enum OriginateProblem { Prefix(#[from] bgp_primitives::ip_prefix::PrefixError), #[error("origin {0:?} must be IGP, EGP or INCOMPLETE")] Origin(String), - #[error("as_path_prepend must be at most {max}")] - Prepend { max: u32 }, + #[error("as_path_prepend of {value} is more than {max}")] + Prepend { value: u32, max: u32 }, #[error("{0}")] Community(#[from] bgp_packet::community::CommunityError), #[error("listed more than once")] diff --git a/crates/server/src/export.rs b/crates/server/src/export.rs index 7722f09..52083bf 100644 --- a/crates/server/src/export.rs +++ b/crates/server/src/export.rs @@ -410,7 +410,10 @@ pub(crate) fn withdrawals( routes: &[Nlri], towards: &Towards, ) -> Vec { - let at_once = (UPDATE_ROOM - 16) / nlri_octets(version); + // What the message needs besides the prefixes: its two length fields, + // or the header of MP_UNREACH_NLRI; 16 octets is more than either. + const WITHDRAWAL_OVERHEAD: usize = 16; + let at_once = (UPDATE_ROOM - WITHDRAWAL_OVERHEAD) / nlri_octets(version); // IPv4 routes that went in MP_REACH_NLRI are withdrawn the same way. let in_body = version == IpVersion::Ipv4 && !(towards.extended_next_hop && towards.next_hop.is_ipv6()); @@ -1076,5 +1079,30 @@ mod tests { routes.len() ); assert!(gone.len() < 8); + let in_mp_unreach = |update: &UpdateMessage, count: usize| { + update.withdrawn_routes.is_empty() + && update.path_attributes.iter().any(|attribute| { + matches!(attribute, PathAttribute::MpUnreachNlri(mp) if mp.prefixes.len() == count) + }) + }; + let v6_routes: Vec = (0..3) + .map(|n| { + IpPrefix::try_from(format!("2001:db8:{n}::/48").as_str()) + .unwrap() + .into() + }) + .collect(); + let gone = withdrawals(IpVersion::Ipv6, &v6_routes, &narrow); + assert_eq!(gone.len(), 1); + assert!(in_mp_unreach(&gone[0], 3), "{:?}", gone[0]); + // IPv4 routes that went with IPv6 next hops (RFC 8950) too. + let over_v6 = Towards { + extended_next_hop: true, + next_hop: "2001:db8::2".parse::().unwrap().into(), + ..narrow + }; + let gone = withdrawals(IpVersion::Ipv4, &routes[..2], &over_v6); + assert_eq!(gone.len(), 1); + assert!(in_mp_unreach(&gone[0], 2), "{:?}", gone[0]); } } diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index 142f774..0902c68 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -477,13 +477,9 @@ impl Fsm { error, SessionError::HoldTimerExpired | SessionError::Collision ); - let notification = error.notification().map(|notification| { - if for_good && self.notifications_restart() { - restart::hard_reset(notification) - } else { - notification - } - }); + let notification = error + .notification() + .map(|notification| self.as_sent(notification, for_good)); let mut actions: Vec = notification .map(|notification| Action::Send(Message::Notification(notification))) .into_iter() @@ -520,16 +516,23 @@ impl Fsm { error: BgpError::Cease(CeaseSubcode::AdminShutdown), data: vec![], }; - let notification = if self.notifications_restart() { - restart::hard_reset(shutdown) - } else { - shutdown - }; + let notification = self.as_sent(shutdown, true); let mut actions = vec![Action::Send(Message::Notification(notification))]; actions.extend(self.close(Ok(()))); actions } + /// `notification` as it goes to the peer: where NOTIFICATIONs count as + /// restarts (RFC 8538) and this one is `for_good`, as a Hard Reset, so + /// that the peer drops our routes rather than keeping them. + fn as_sent(&self, notification: NotificationMessage, for_good: bool) -> NotificationMessage { + if for_good && self.notifications_restart() { + restart::hard_reset(notification) + } else { + notification + } + } + fn close(&mut self, result: Result<(), SessionError>) -> Vec { self.phase = Phase::Closed; vec![Action::Close(result)] diff --git a/crates/server/src/policy.rs b/crates/server/src/policy.rs index 03fd158..957842a 100644 --- a/crates/server/src/policy.rs +++ b/crates/server/src/policy.rs @@ -855,9 +855,9 @@ fn is_listed_bogon(prefix: &IpPrefix) -> bool { BOGONS.iter().any(|bogon| bogon.contains(prefix)) } -/// A list of prefixes a peer may announce. Each entry allows the prefix -/// itself and, if it says so, the more specific prefixes inside it up to a -/// length. +/// A set of prefixes: what a peer may announce, what is our own, what may +/// be announced to a peer. Each entry allows the prefix itself and, if it +/// says so, the more specific prefixes inside it up to a length. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct PrefixSet { /// For each prefix listed, the longest prefix inside it that is @@ -880,10 +880,6 @@ impl PrefixSet { } } - pub fn len(&self) -> usize { - self.entries.len() - } - pub fn is_empty(&self) -> bool { self.entries.is_empty() } @@ -1372,6 +1368,17 @@ mod tests { assert!(!Peer.may_be_sent_only_to_customer_routes()); } + /// AS_TRANS (23456) is a bogon where real numbers fit, and stands in + /// for one that does not on a session with two-octet AS numbers. + #[test] + fn as_trans_is_a_bogon_only_where_wide_numbers_fit() { + assert!(is_bogon_asn(u32::from(AS_TRANS), true)); + assert!(!is_bogon_asn(u32::from(AS_TRANS), false)); + assert!(is_bogon_asn(0, false)); + assert!(is_bogon_asn(64512, false)); + assert!(!is_bogon_asn(3333, false)); + } + /// Export limits are for networks above or beside us, and for one we /// know nothing of; not for customers, route server clients or our own /// routers, which are told of what they are to be. diff --git a/crates/server/src/restart.rs b/crates/server/src/restart.rs index 9ba635f..5554e62 100644 --- a/crates/server/src/restart.rs +++ b/crates/server/src/restart.rs @@ -28,8 +28,10 @@ use bgp_packet::message::NotificationMessage; use bgp_packet::open::{Capability, GracefulRestartEntry, OpenMessage}; use bgp_primitives::constants::{AfiSafi, IpVersion}; use bgp_rib::table::{PeerIndex, RibTable}; +use tracing::{error, info}; + +use crate::connection::off_thread; use tokio::sync::watch; -use tracing::info; use crate::address::PeerAddress; use crate::error::SessionError; @@ -343,20 +345,21 @@ impl Restarts { let Some(kept) = kept else { return; }; - let rib = Arc::clone(&self.rib); - let outcome = tokio::task::spawn_blocking(move || { - let outcome = rib.remove_peer(kept.rib_peer); - rib.attrs().sweep(); - outcome - }) - .await - .unwrap_or_default(); - info!( - peer = %peer.0, - peer_asn = peer.1, - removed_routes = outcome.removed, - "The peer did not come back in time; removed the routes kept while it restarted" - ); + let rib_peer = kept.rib_peer; + match off_thread(&self.rib, move |table| table.remove_peer(rib_peer)).await { + Ok(outcome) => info!( + peer = %peer.0, + peer_asn = peer.1, + removed_routes = outcome.removed, + "The peer did not come back in time; removed the routes kept while it restarted" + ), + Err(e) => error!( + peer = %peer.0, + peer_asn = peer.1, + err = %e, + "The peer did not come back in time, and removing the routes kept for it failed" + ), + } } /// The routes kept of `peer`, for its new session to take over. @@ -445,3 +448,102 @@ pub(crate) fn ends_in_restart(result: &Result<(), SessionError>, agreed: &Agreed _ => false, } } + +#[cfg(test)] +mod tests { + use std::io; + + use super::*; + + fn agreed(notification: bool) -> Agreed { + Agreed { + restart_time: 120, + restarted: false, + notification, + families: FamilySet::supported(), + forwarding: FamilySet::default(), + } + } + + /// Which ends of a session count as the peer restarting (RFC 4724 + /// §4.2, RFC 8538): a connection that drops always; a NOTIFICATION, + /// or an expired hold timer, only where both sides agreed that they + /// do, and never a Hard Reset. + #[test] + fn a_session_ends_in_a_restart_as_agreed() { + let dropped: Result<(), SessionError> = + Err(io::Error::from(io::ErrorKind::BrokenPipe).into()); + let hold_timer = Err(SessionError::HoldTimerExpired); + let notified = Err(SessionError::NotifiedByPeer(BgpError::Cease( + CeaseSubcode::AdminReset, + ))); + let hard_reset = Err(SessionError::NotifiedByPeer(BgpError::Cease( + CeaseSubcode::HardReset, + ))); + let ours = Err(SessionError::PrefixLimitExceeded { limit: 1 }); + for (result, plain, with_notifications) in [ + (&dropped, true, true), + (&hold_timer, false, true), + (¬ified, false, true), + (&hard_reset, false, false), + (&ours, false, false), + (&Ok(()), false, false), + ] { + assert_eq!(ends_in_restart(result, &agreed(false)), plain, "{result:?}"); + assert_eq!( + ends_in_restart(result, &agreed(true)), + with_notifications, + "{result:?} with notifications" + ); + } + } + + /// A Hard Reset carries the NOTIFICATION it stands in for in its data, + /// and one that is a Hard Reset already is left alone. + #[test] + fn a_hard_reset_wraps_the_notification() { + let shutdown = NotificationMessage { + error: BgpError::Cease(CeaseSubcode::AdminShutdown), + data: vec![7], + }; + let reset = hard_reset(shutdown); + assert_eq!(reset.error, BgpError::Cease(CeaseSubcode::HardReset)); + assert_eq!(reset.data, [6, 2, 7]); + assert_eq!(hard_reset(reset.clone()), reset); + } + + /// The configuration's graceful restart settings: what is given + /// overrides the default, a restart time has to fit its twelve bits, + /// and `disabled` is the opposite of `enabled`. + #[test] + fn graceful_restart_settings_override_the_defaults() { + let proto = |text: &str| { + protobuf::text_format::parse_from_str::(text) + .unwrap() + }; + let defaults = GracefulRestartConfig::default(); + let same = defaults.overridden_by(&proto("")).unwrap(); + assert_eq!(same, defaults); + let changed = defaults + .overridden_by(&proto( + "disabled: true restart_time_seconds: 30 stale_routes_seconds: 10", + )) + .unwrap(); + assert!(!changed.enabled); + assert_eq!(changed.restart_time, 30); + assert_eq!(changed.stale_routes, Duration::from_secs(10)); + assert_eq!( + defaults + .overridden_by(&proto("restart_time_seconds: 4095")) + .unwrap() + .restart_time, + MAX_RESTART_TIME + ); + assert_eq!( + defaults + .overridden_by(&proto("restart_time_seconds: 4096")) + .unwrap_err(), + crate::error::GracefulRestartProblem::RestartTime(4096) + ); + } +} diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index 2a7eeee..7c57519 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -22,7 +22,7 @@ use crate::address::{PeerAddress, socket_text}; use crate::advertise::spawn_advertiser; use crate::collision::{CollisionTable, Direction}; use crate::config::{Config, PeerConfig}; -use crate::connection::{PeerSettings, ServerContext, run_session}; +use crate::connection::{PeerSettings, ServerContext, off_thread, run_session}; use crate::dialer::spawn_dialers; use crate::error::ServerError; use crate::metrics::{CONNECTIONS_REFUSED, PEER_MAX_PREFIXES}; @@ -361,17 +361,20 @@ pub async fn leave_for_restart(registry: &SessionRegistry, kernel: Option<&Fib>) { warn!("Letting go of the kernel's routes took too long"); } - let left = registry.leave(Leave::Restart, SHUTDOWN_GRACE).await; - if left > 0 { - warn!(sessions = left, "Some sessions did not close in time"); - } + leave_sessions(registry, Leave::Restart).await; } /// Closes every session politely and waits a moment for them to go: each /// peer is told we are going (Cease / Administrative Shutdown), so that it /// stops using our routes now and not when its hold timer expires. pub async fn close_sessions(registry: &SessionRegistry) { - let left = registry.shutdown(SHUTDOWN_GRACE).await; + leave_sessions(registry, Leave::Stop).await; +} + +/// Closes every session as `leave` says, and waits a moment for them to +/// go. +async fn leave_sessions(registry: &SessionRegistry, leave: Leave) { + let left = registry.leave(leave, SHUTDOWN_GRACE).await; if left > 0 { warn!(sessions = left, "Some sessions did not close in time"); } @@ -506,7 +509,9 @@ enum Refusal { NotAPeer, /// Its address has as many connections open as one may. TooMany, - /// The peer went over its limit of routes and is being kept down. + /// The peer is being kept down: it went over its limit of routes, we + /// were to send it more than its export limit allows, or its sessions + /// keep failing. HeldDown, } @@ -603,10 +608,11 @@ fn spawn_sweeper(rib: Arc) -> JoinHandle<()> { if !piled_up && swept_at.elapsed() < SWEEP_INTERVAL { continue; } - let table = Arc::clone(&rib); // Takes each of the store's locks in turn and looks at every // set: not for a thread that sessions run on. - let _ = tokio::task::spawn_blocking(move || table.attrs().sweep()).await; + if let Err(e) = off_thread(&rib, |_| ()).await { + error!(err = %e, "Sweeping the attribute sets failed"); + } swept_at = Instant::now(); kept = rib.attrs().len(); } diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index 9bf57be..b184055 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -1,4 +1,7 @@ -//! Registry of established BGP sessions. +//! The sessions as the rest of the server sees them: the registry of +//! those that are established, the configured peers and where the dialer +//! is with each, the events a session's coming and going raise, and the +//! word that the daemon is leaving. //! //! A session registers itself when it reaches Established and is removed when //! its [`SessionGuard`] drops, so a session that errors out is always @@ -389,6 +392,7 @@ impl SessionRegistry { if let Some(name) = &handle.name { PEER_UP.with_label_values(&[name]).set(1); } + // Fails only when nobody is listening, which is fine. let _ = self.events.send(SessionEvent::Up(handle)); SessionGuard { registry: self.clone(), @@ -409,7 +413,7 @@ impl SessionRegistry { } } - /// True once [`shutdown`](Self::shutdown) has begun; dialers stop + /// True once [`leave`](Self::leave) has been called; dialers stop /// reconnecting. pub fn is_shutting_down(&self) -> bool { self.shutting_down.load(Ordering::Relaxed) @@ -419,10 +423,6 @@ impl SessionRegistry { /// 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. - pub async fn shutdown(&self, grace: Duration) -> usize { - self.leave(Leave::Stop, grace).await - } - /// 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.