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

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());
 

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

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<Control>,
+        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,

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

crates/server/src/config.rs

@@ -119,10 +119,7 @@ impl Config {
     }
 
     pub fn from_proto(proto: BgpConfig) -> Result<Self, ConfigError> {
-        // 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> {
+    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> {
+    u8::try_from(prepend)
+        .ok()
+        .filter(|prepend| u32::from(*prepend) <= MAX_AS_PATH_PREPEND)
+}
+
 fn parse_export(export: &proto::ExportPolicy) -> Result<ExportPolicy, ExportProblem> {
     Ok(ExportPolicy {
         learned_routes: export.learned_routes,
@@ -326,10 +337,10 @@ fn parse_export(export: &proto::ExportPolicy) -> Result<ExportPolicy, ExportProb
             &export.add_communities,
             &export.add_large_communities,
         )?,
-        prepend: u8::try_from(export.as_path_prepend)
-            .ok()
-            .filter(|prepend| u32::from(*prepend) <= MAX_AS_PATH_PREPEND)
-            .ok_or(ExportProblem::Prepend(export.as_path_prepend))?,
+        prepend: parse_prepend(export.as_path_prepend).ok_or(ExportProblem::Prepend {
+            value: export.as_path_prepend,
+            max: MAX_AS_PATH_PREPEND,
+        })?,
     })
 }
 
@@ -620,10 +631,8 @@ fn parse_originated_route(route: &OriginatedRoute) -> Result<LocalRoute, Origina
         "INCOMPLETE" => 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 {
+    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
@@ -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!(

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<FamilySet>,
+    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<T: Send + 'static>(
+    table: &Arc<RibTable>,
+    work: impl FnOnce(&RibTable) -> T + Send + 'static,
+) -> Result<T, tokio::task::JoinError> {
+    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<Established>) -> Wake {
     let Some(Established {
         outbound,

crates/server/src/dialer.rs

@@ -121,7 +121,11 @@ async fn dial_loop(peer: PeerConfig, server: Arc<ServerContext>, 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<ServerContext>, 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");
+    }
+}

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")]

crates/server/src/export.rs

@@ -410,7 +410,10 @@ pub(crate) fn withdrawals(
     routes: &[Nlri],
     towards: &Towards,
 ) -> Vec<UpdateMessage> {
-    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<Nlri> = (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::<std::net::Ipv6Addr>().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]);
     }
 }

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<Action> = 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<Action> {
         self.phase = Phase::Closed;
         vec![Action::Close(result)]

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.

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!(
+        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),
+            (&notified, 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::<crate::proto::config::GracefulRestart>(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)
+        );
+    }
+}

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<RibTable>) -> 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();
         }

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.