session: the way out to the peer as a type of its own

A code-quality pass over the per-peer session. The outgoing side of a
session, eight fields served from six places in a two-thousand-line
file, is `Outgoing`: the write buffer, the batches waiting for the
writer task, the task and its word on why it ended. One real race
went with it: when the writer task failed with the queue full, the
session took whichever of two ready branches the select chose, and
half the time ended with an invented send hold timer expiry instead
of the connection's error, which decides whether the peer's routes
are kept over a restart; the writer's own word is taken now. The run
loop's handling of what the rest of the server has to send is a
method that goes by the state machine's state, so that pieces after a
NOTIFICATION in the same batch do not follow it out; `execute_all`
reads one line an action with the sending, delivering and
after-failure steps named. A peer's routes have one way onto the
table at a session's end, whether removed or kept for a restart. An
UPDATE whose attributes cannot be kept ends the session under an
error of its own, `Unstorable`, where a message of ours that cannot
be encoded is dropped. What is kept of a restarting peer is a value,
not seven arguments. The dialer no longer turns a session's end into
a Result and back. The server's screening of a connection is called
that, apart from the collision check's `admit`.

Testing: the back-pressure machinery has unit tests at last: batches
wait while the writer has no room, room goes to the oldest batch, a
piece is counted as the messages in it, and a lost connection is
reported as one while batches wait. The routes module gains tests of
holding at its bound, forgetting and revalidation. The server tests
and the whole integration suite pass.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

crates/server/src/dialer.rs

@@ -18,7 +18,7 @@ use tracing::{info, warn};
 
 use crate::collision::Direction;
 use crate::config::PeerConfig;
-use crate::connection::{CLOSED_AT_OUR_REQUEST, PeerSettings, ServerContext, run_session};
+use crate::connection::{PeerSettings, ServerContext, run_session};
 use crate::session::DialState;
 
 /// Upper bound for exponential backoff between connection attempts.
@@ -102,7 +102,9 @@ async fn dial_loop(peer: PeerConfig, server: Arc<ServerContext>, connect_retry:
         registry.set_dial_state(&peer.name, DialState::Connecting);
         info!(peer = %peer.name, endpoint = %peer.endpoint, "Connecting to peer");
 
-        let ended = match tokio::time::timeout(CONNECT_TIMEOUT, connect(&peer)).await {
+        // Why the attempt ended, and whether that is a failure worth a
+        // line in the log.
+        let (reason, failed) = match tokio::time::timeout(CONNECT_TIMEOUT, connect(&peer)).await {
             Ok(Ok(stream)) => match stream.peer_addr() {
                 Ok(peer_addr) => {
                     info!(peer = %peer.name, endpoint = %peer.endpoint, "TCP connected, starting session");
@@ -121,25 +123,17 @@ async fn dial_loop(peer: PeerConfig, server: Arc<ServerContext>, connect_retry:
                     if end.established {
                         backoff = connect_retry;
                     }
-                    if end.result.is_ok() {
-                        Ok(())
-                    } else {
-                        Err(end.reason())
+                    let failed = end.result.is_err();
+                    (end.reason(), failed)
                 }
-                }
-                Err(e) => Err(format!("connection lost at once: {e}")),
+                Err(e) => (format!("connection lost at once: {e}"), true),
             },
-            Ok(Err(e)) => Err(format!("connect failed: {e}")),
-            Err(_) => Err(format!("connect timed out after {CONNECT_TIMEOUT:?}")),
+            Ok(Err(e)) => (format!("connect failed: {e}"), true),
+            Err(_) => (format!("connect timed out after {CONNECT_TIMEOUT:?}"), true),
         };
-
-        let reason = match ended {
-            Ok(()) => CLOSED_AT_OUR_REQUEST.to_string(),
-            Err(reason) => {
+        if failed {
             warn!(peer = %peer.name, endpoint = %peer.endpoint, err = %reason, "Session attempt ended");
-                reason
         }
-        };
 
         // Not all at the same moment, when many peers went away together.
         let wait = jittered(backoff);

crates/server/src/error.rs

@@ -41,6 +41,9 @@ pub enum SessionError {
     Decode(#[from] DecodeError),
     #[error("could not encode a message: {0}")]
     Encode(#[from] EncodeError),
+    /// A received UPDATE's attributes could not be kept in the table.
+    #[error(transparent)]
+    Unstorable(#[from] crate::routes::Unstorable),
     /// The peer sent nothing for as long as the pre-negotiation hold time.
     #[error("timed out waiting for a message in the {0} state")]
     Timeout(State),
@@ -101,9 +104,10 @@ impl SessionError {
     pub fn notification(&self) -> Option<NotificationMessage> {
         let error = match self {
             Self::Io(_) | Self::SendHoldTimerExpired(_) | Self::NotifiedByPeer(_) => return None,
-            // Something of ours could not be written out, which is a bug
-            // here: as for a panic, the session is reset.
-            Self::Encode(_) => BgpError::Cease(CeaseSubcode::AdminReset),
+            // Something of ours could not be written out, or the peer's
+            // routes could not be kept, which is a bug here: as for a
+            // panic, the session is reset.
+            Self::Encode(_) | Self::Unstorable(_) => BgpError::Cease(CeaseSubcode::AdminReset),
             Self::Decode(e) => {
                 return Some(NotificationMessage {
                     error: e.notification.clone(),

crates/server/src/restart.rs

@@ -244,6 +244,30 @@ pub struct Kept {
     pub until: SystemTime,
 }
 
+impl Kept {
+    /// What was kept of a peer: `routes` routes of the IP versions
+    /// `versions`, under `rib_peer`, of the peer at `addr` called `name`.
+    /// Which restart they are of and until when they are kept are the
+    /// table's to say, in [`Restarts::keep`].
+    pub(crate) fn new(
+        addr: SocketAddr,
+        name: Option<String>,
+        rib_peer: PeerIndex,
+        versions: Vec<IpVersion>,
+        routes: usize,
+    ) -> Self {
+        Self {
+            id: 0,
+            addr,
+            rib_peer,
+            name,
+            versions,
+            routes,
+            until: SystemTime::UNIX_EPOCH,
+        }
+    }
+}
+
 /// What there is of graceful restart while the server runs: the routes
 /// kept of peers that are restarting, and how far the peers are with
 /// sending their routes after this daemon started.
@@ -298,33 +322,14 @@ impl Restarts {
         })
     }
 
-    /// Notes that the routes of `peer` under `rib_peer` have been kept
-    /// (`routes` of them, of the IP versions `versions`) until `for_at_most`
-    /// from now, and has them removed then unless the peer is back.
-    #[allow(clippy::too_many_arguments)]
-    pub(crate) fn keep(
-        self: &Arc<Self>,
-        peer: PeerKey,
-        addr: SocketAddr,
-        name: Option<String>,
-        rib_peer: PeerIndex,
-        versions: Vec<IpVersion>,
-        routes: usize,
-        for_at_most: Duration,
-    ) {
+    /// Notes that `kept` says what routes of `peer` have been kept, until
+    /// `for_at_most` from now, and has them removed then unless the peer
+    /// is back.
+    pub(crate) fn keep(self: &Arc<Self>, peer: PeerKey, mut kept: Kept, for_at_most: Duration) {
         let id = self.ids.fetch_add(1, Ordering::Relaxed);
-        self.kept.locked().insert(
-            peer,
-            Kept {
-                id,
-                addr,
-                rib_peer,
-                name,
-                versions,
-                routes,
-                until: SystemTime::now() + for_at_most,
-            },
-        );
+        kept.id = id;
+        kept.until = SystemTime::now() + for_at_most;
+        self.kept.locked().insert(peer, kept);
         let restarts = Arc::clone(self);
         tokio::spawn(async move {
             tokio::time::sleep(for_at_most).await;

crates/server/src/routes.rs

@@ -21,6 +21,14 @@ use crate::rpki::Checks;
 /// than a peer sending the whole Internet has; a bound all the same.
 const MAX_HELD: usize = 200_000;
 
+/// Why an UPDATE could not be applied: the attributes its routes were to
+/// be kept with could not be written in the form the table keeps them in.
+/// Attributes that were decoded from a message always can be, so this is
+/// a bug here, and not anything the peer did.
+#[derive(Debug, thiserror::Error)]
+#[error("could not keep the routes' attributes: {0}")]
+pub struct Unstorable(#[from] pub(crate) EncodeError);
+
 /// What applying an UPDATE did.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub(crate) struct Applied {
@@ -113,7 +121,7 @@ pub(crate) fn apply(
     held: &mut Held,
     update: &UpdateMessage,
     attributes_wire: Option<&[u8]>,
-) -> Result<Applied, EncodeError> {
+) -> Result<Applied, Unstorable> {
     let mut applying = Applying {
         table,
         import,
@@ -598,3 +606,254 @@ pub(crate) fn revalidate(
     }
     applied
 }
+
+#[cfg(test)]
+mod tests {
+    use std::net::{IpAddr, Ipv4Addr};
+
+    use bgp_packet::message::{AsPathAttribute, NextHopPathAttribute, OriginPathAttribute};
+    use bgp_rib::table::PeerInfo;
+    use bgp_rpki::{Vrp, VrpTable};
+
+    use super::*;
+    use crate::rejections::RejectionLog;
+
+    const PEER_ASN: u32 = 65001;
+    const PEER_ADDRESS: Ipv4Addr = Ipv4Addr::new(192, 0, 2, 1);
+
+    fn prefix(text: &str) -> IpPrefix {
+        IpPrefix::try_from(text).unwrap()
+    }
+
+    fn nlri(text: &str) -> Nlri {
+        prefix(text).into()
+    }
+
+    fn attrs_of(route: &str) -> Arc<AttrSet> {
+        RibTable::new(1)
+            .attrs()
+            .intern(&announcement(route).path_attributes, None)
+            .unwrap()
+    }
+
+    /// An UPDATE as the peer announces `route`, originating it itself.
+    fn announcement(route: &str) -> UpdateMessage {
+        UpdateMessage {
+            withdrawn_routes: vec![],
+            path_attributes: vec![
+                PathAttribute::Origin(OriginPathAttribute::IGP),
+                AsPathAttribute::from_asns(vec![PEER_ASN]),
+                PathAttribute::NextHop(NextHopPathAttribute(PEER_ADDRESS)),
+            ],
+            announced_routes: vec![nlri(route)],
+        }
+    }
+
+    /// VRPs that allow `route` to be originated by `asn` and nobody else.
+    fn vrps(route: &str, asn: u32) -> VrpTable {
+        let prefix = prefix(route);
+        VrpTable::new([Vrp {
+            prefix,
+            max_length: prefix.length(),
+            asn,
+        }])
+    }
+
+    /// A table with the peer in it, and what is needed to apply its
+    /// UPDATEs by a permissive policy, checking only against `vrps`.
+    struct Fixture {
+        table: RibTable,
+        peer: PeerIndex,
+        policy: ImportPolicy,
+        rejections: RejectionLog,
+    }
+
+    impl Fixture {
+        fn new() -> Self {
+            let table = RibTable::new(1);
+            let peer = table
+                .add_peer(PeerInfo {
+                    asn: PEER_ASN,
+                    router_id: Ipv4Addr::new(10, 0, 0, 2),
+                    address: PEER_ADDRESS.into(),
+                    interface: 0,
+                    ibgp: false,
+                    reflector_client: false,
+                })
+                .unwrap();
+            Self {
+                table,
+                peer,
+                policy: ImportPolicy::permissive(),
+                rejections: RejectionLog::default(),
+            }
+        }
+
+        fn with_import<T>(
+            &self,
+            vrps: Option<&VrpTable>,
+            held: &mut Held,
+            work: impl FnOnce(&RibTable, &Import<'_>, &mut Held) -> T,
+        ) -> T {
+            let import = Import {
+                peer: self.peer,
+                from: RouteSource {
+                    peer_asn: PEER_ASN,
+                    local_asn: 65000,
+                    ibgp: false,
+                    four_octet_asn: true,
+                    role: None,
+                    peer_address: PEER_ADDRESS.into(),
+                    local_address: Ipv4Addr::new(192, 0, 2, 9).into(),
+                    extended_next_hop: false,
+                },
+                policy: &self.policy,
+                checks: Checks {
+                    vrps,
+                    ..Checks::NONE
+                },
+                rejections: &self
+                    .rejections
+                    .of_peer(IpAddr::from(PEER_ADDRESS).into(), "test"),
+            };
+            work(&self.table, &import, held)
+        }
+
+        fn announce(&self, vrps: Option<&VrpTable>, held: &mut Held, route: &str) -> Applied {
+            self.with_import(vrps, held, |table, import, held| {
+                apply(table, import, held, &announcement(route), None).unwrap()
+            })
+        }
+
+        fn revalidate(
+            &self,
+            vrps: Option<&VrpTable>,
+            held: &mut Held,
+            suspects: &[Nlri],
+        ) -> Applied {
+            self.with_import(vrps, held, |table, import, held| {
+                revalidate(table, import, held, suspects)
+            })
+        }
+    }
+
+    /// Past the bound, a new route is dropped and the overflow noted, so
+    /// that the peer can be asked for it later; one already held is still
+    /// replaced, since that takes no more room.
+    #[test]
+    fn holding_stops_at_the_bound_and_says_so() {
+        let mut held = Held::default();
+        let attrs = attrs_of("198.51.100.0/24");
+        for n in 0..MAX_HELD as u32 {
+            let [a, b, c, d] = (0x0a00_0000 + n).to_be_bytes();
+            held.hold(
+                IpPrefix::new(IpVersion::Ipv4, [a, b, c, d], 32)
+                    .unwrap()
+                    .into(),
+                &attrs,
+            );
+        }
+        assert_eq!(held.len(), MAX_HELD);
+        assert!(!held.take_overflowed());
+
+        held.hold(nlri("198.51.100.0/24"), &attrs);
+        assert_eq!(held.len(), MAX_HELD);
+        assert!(held.take_overflowed());
+        assert!(
+            !held.take_overflowed(),
+            "noted once, until it happens again"
+        );
+
+        let other = attrs_of("203.0.113.0/24");
+        held.hold(nlri("10.0.0.0/32"), &other);
+        assert_eq!(held.len(), MAX_HELD);
+        assert!(!held.take_overflowed());
+        assert!(Arc::ptr_eq(&held.routes[&nlri("10.0.0.0/32")], &other));
+    }
+
+    #[test]
+    fn forgetting_takes_out_what_is_named_and_nothing_else() {
+        let mut held = Held::default();
+        let attrs = attrs_of("198.51.100.0/24");
+        held.hold(nlri("198.51.100.0/24"), &attrs);
+        held.hold(nlri("203.0.113.0/24"), &attrs);
+
+        held.forget(&[nlri("198.51.100.0/24"), nlri("192.0.2.0/24")]);
+        assert_eq!(held.len(), 1);
+        assert!(held.routes.contains_key(&nlri("203.0.113.0/24")));
+
+        held.forget(&[nlri("203.0.113.0/24")]);
+        held.forget(&[nlri("203.0.113.0/24")]);
+        assert_eq!(held.len(), 0);
+    }
+
+    /// A route the VRPs find invalid stays out of the table but is not
+    /// lost: when the VRPs change in its favour it goes in, without the
+    /// peer having to send it again.
+    #[test]
+    fn a_held_route_is_released_once_the_checks_pass_it() {
+        let fixture = Fixture::new();
+        let mut held = Held::default();
+        let route = "198.51.100.0/24";
+        let against = vrps(route, PEER_ASN + 1);
+        let applied = fixture.announce(Some(&against), &mut held, route);
+        assert_eq!(applied.rejected, 1);
+        assert_eq!(applied.outcome.added, 0);
+        assert!(fixture.table.is_empty());
+        assert_eq!(held.len(), 1);
+
+        // Nothing has changed in the route's favour yet.
+        let applied = fixture.revalidate(Some(&against), &mut held, &[]);
+        assert_eq!(applied, Applied::default());
+        assert_eq!(held.len(), 1);
+
+        let for_it = vrps(route, PEER_ASN);
+        let applied = fixture.revalidate(Some(&for_it), &mut held, &[]);
+        assert_eq!(applied.outcome.added, 1);
+        assert_eq!(applied.rejected, 0);
+        assert_eq!(held.len(), 0);
+        assert_eq!(fixture.table.routes(&prefix(route)).len(), 1);
+    }
+
+    /// The other way round: a route in the table that the VRPs come to
+    /// find invalid is taken out and held, and comes back once they
+    /// change again.
+    #[test]
+    fn a_route_turned_invalid_is_held_and_comes_back() {
+        let fixture = Fixture::new();
+        let mut held = Held::default();
+        let route = "198.51.100.0/24";
+        let for_it = vrps(route, PEER_ASN);
+        let applied = fixture.announce(Some(&for_it), &mut held, route);
+        assert_eq!(applied.outcome.added, 1);
+        assert_eq!(fixture.table.len(), 1);
+
+        let against = vrps(route, PEER_ASN + 1);
+        let applied = fixture.revalidate(Some(&against), &mut held, &[nlri(route)]);
+        assert_eq!(applied.rejected, 1);
+        assert_eq!(applied.outcome.removed, 1);
+        assert!(fixture.table.is_empty());
+        assert_eq!(held.len(), 1);
+
+        let applied = fixture.revalidate(Some(&for_it), &mut held, &[]);
+        assert_eq!(applied.outcome.added, 1);
+        assert_eq!(held.len(), 0);
+        assert_eq!(fixture.table.len(), 1);
+    }
+
+    /// Without anything to check against there is nothing to look at
+    /// again: what is held stays held, and what is in the table stays.
+    #[test]
+    fn revalidation_without_checks_changes_nothing() {
+        let fixture = Fixture::new();
+        let mut held = Held::default();
+        let route = "198.51.100.0/24";
+        fixture.announce(Some(&vrps(route, PEER_ASN + 1)), &mut held, route);
+        assert_eq!(held.len(), 1);
+
+        let applied = fixture.revalidate(None, &mut held, &[nlri(route)]);
+        assert_eq!(applied, Applied::default());
+        assert_eq!(held.len(), 1);
+        assert!(fixture.table.is_empty());
+    }
+}

crates/server/src/server.rs

@@ -237,7 +237,7 @@ impl BgpServer {
             };
             match result {
                 Ok((stream, peer_addr)) => {
-                    match admit(&self.config, &context, peer_addr.into(), &connections) {
+                    match screen(&self.config, &context, peer_addr.into(), &connections) {
                         Ok((peer, slot)) => {
                             let context = Arc::clone(&context);
                             tokio::spawn(serve(stream, peer_addr, context, peer, slot));
@@ -461,9 +461,10 @@ fn listen(addr: SocketAddr) -> io::Result<TcpListener> {
     socket.listen(LISTEN_BACKLOG)
 }
 
-/// Decides whether to take a connection from `address`, and if so with what
-/// settings.
-fn admit(
+/// Screens a connection from `address`: whether to take it, and if so
+/// with what settings. Not the collision check, which is
+/// [`crate::collision::CollisionTable::admit`].
+fn screen(
     config: &Config,
     context: &ServerContext,
     address: PeerAddress,