Review pass on the server: bugs fixed, session setup simplified

A review of the session driver, the server loop, the dialer, the
advertiser and the HTTP API, with the fixes.

Bugs:

- A peer that stopped reading held its session, and our routes, for as
  long as TCP kept retrying: a blocked write was awaited with no timer
  running. Writes now give up after the Send Hold Time of RFC 9687.
- Routes we originate were sent to a peer in our own AS with our AS
  number on the path and no LOCAL_PREF, which such a peer refuses. They
  now go with an empty path and a LOCAL_PREF.
- Collision detection knew a peer by its BGP Identifier alone. A
  neighbour with one session over IPv4 and one over IPv6 had the second
  closed as a collision, and any host that could connect could displace
  a peer's connection by claiming its identifier. A peer is now its
  address and identifier together.
- The dialer reset its backoff whenever TCP connected, so a peer that
  accepts the connection and refuses the session was dialled every few
  seconds for ever. It resets only after a session was established.
- A failure on our side (routes that cannot be stored) dropped the
  connection without a NOTIFICATION. It goes through the state machine
  and the peer is sent Cease. A queued message that cannot be encoded
  is logged and skipped instead of ending the session.
- The accept loop retried at once after an error, spinning when the
  process was out of file descriptors.
- Attribute sets no route used any more were dropped only when a peer
  went away. A task now does it every minute.
- The advertiser subscribed to session events inside its task, after the
  dialers were started, and could miss the first session.
- Shutdown waited without limit on a session whose queue was full.
- /api/routes/received took any limit and built the reply on a runtime
  thread. The limit is capped and the work is done off those threads.
- The default log level was TRACE, which logs every UPDATE.
- IPv6 listeners are IPv6-only: listening on 0.0.0.0 and [::] fails on
  systems where the second would take IPv4 too, and IPv4 peers arriving
  under mapped addresses matched no configured peer.
- A connection closed after a NOTIFICATION is kept for a moment so that
  the peer can read it.
- The configuration accepted two peers with one name or one address,
  and a router id of 0.0.0.0.
- An HTTP address that cannot be bound now stops the daemon at startup.

Simplifications:

- SessionConfig mixed what a server's sessions share with what is
  configured for one peer, and was cloned and patched field by field in
  two places. It is ServerContext, shared, and PeerSettings, made from a
  PeerConfig in one place.
- The attributes of an UPDATE as they were on the wire travelled from
  the reader to the table through a field of the session, around the
  state machine. They go through it, with the message.
- run_session returns how the session ended and whether it had been
  established; Ok no longer reads as "closed by peer".
- Session is not generic over its reader.
- BgpServerBuilder had nothing to build: BgpServer::bind takes the
  configuration.
- One shutdown path, close_sessions, for the signal handler and for a
  failed server.
- The advertiser's three copies of "send every route" are one.
- PeerId moved from the rib crate, which did not use it, to the server.
- Unused API removed or made private to the crate.

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

Cargo.lock

@@ -262,6 +262,7 @@ dependencies = [
  "proptest",
  "protobuf",
  "protobuf-codegen",
+ "socket2",
  "thiserror 2.0.12",
  "tokio",
  "tracing",

Cargo.toml

@@ -41,6 +41,7 @@ bytes = "1.10.1"
 eyre = "0.6.12"
 nom = "8.0.0"
 serde = { version = "1.0.219", features = ["derive"] }
+socket2 = "0.6"
 serde_json = "1.0.140"
 serde_repr = "0.1.20"
 thiserror = "2.0.12"

bin/bgp/src/main.rs

@@ -1,14 +1,13 @@
 use std::sync::Arc;
-use std::time::Duration;
 
 use bgp_server::config::Config;
 use bgp_server::proto::config::BgpConfig;
-use bgp_server::server::BgpServerBuilder;
+use bgp_server::server::{BgpServer, close_sessions};
 use bgp_server::tracing::TracingArgs;
 use bgp_web::{AppState, RouterInfo};
 use clap::Parser;
 use eyre::{Context, Result};
-use tracing::{error, info, warn};
+use tracing::{error, info};
 
 #[derive(Parser, Debug)]
 struct ServerArgs {
@@ -31,9 +30,14 @@ async fn main() -> Result<()> {
         .wrap_err("Failed to parse config file")?;
     let config = Config::from_proto(proto_config).wrap_err("Invalid configuration")?;
 
-    let server = BgpServerBuilder::new(config).bind().await?;
+    let server = BgpServer::bind(config).await?;
 
     if let Some(http_endpoint) = server.config().http_endpoint {
+        // Bound here, so that an address that cannot be bound stops the
+        // daemon from starting, as it does for the BGP listeners.
+        let listener = tokio::net::TcpListener::bind(http_endpoint)
+            .await
+            .wrap_err_with(|| format!("Failed to bind HTTP listener on {http_endpoint}"))?;
         let state = AppState {
             info: Arc::new(
                 RouterInfo::new(
@@ -48,7 +52,7 @@ async fn main() -> Result<()> {
             rib: Arc::clone(&server.rib),
         };
         tokio::spawn(async move {
-            if let Err(e) = bgp_web::serve(state, http_endpoint).await {
+            if let Err(e) = bgp_web::serve_on(state, listener).await {
                 error!(err = %e, "HTTP server exited");
             }
         });
@@ -62,10 +66,7 @@ async fn main() -> Result<()> {
         result = server.run() => Ok(result?),
         signal = shutdown_signal() => {
             info!(signal, "Shutting down, closing BGP sessions");
-            let left = registry.shutdown(Duration::from_secs(2)).await;
-            if left > 0 {
-                warn!(sessions = left, "Some sessions did not close in time");
-            }
+            close_sessions(&registry).await;
             Ok(())
         }
     }

crates/integration-tests/src/test_server.rs

@@ -7,7 +7,7 @@ use bgp_rib::LocalRib;
 use bgp_rib::table::{Event, PeerIndex, Reader, RibTable};
 use bgp_server::config::Config;
 use bgp_server::error::ServerError;
-use bgp_server::server::BgpServerBuilder;
+use bgp_server::server::BgpServer;
 use bgp_server::session::SessionRegistry;
 use bgp_web::{AppState, RouterInfo};
 use eyre::Result;
@@ -56,7 +56,7 @@ impl TestServer {
         };
         customize(&mut config);
 
-        let bound = BgpServerBuilder::new(config).bind().await?;
+        let bound = BgpServer::bind(config).await?;
         let addr = bound
             .local_addrs()
             .into_iter()

crates/integration-tests/tests/announce.rs

@@ -8,7 +8,8 @@ use bgp_integration_tests::test_peer::TestPeer;
 use bgp_integration_tests::test_server::{RouteChange, TestServer};
 use bgp_packet::message::{
     AsPathAttribute, CommunitiesPathAttribute, LargeCommunitiesPathAttribute, LargeCommunity,
-    NextHopPathAttribute, OriginPathAttribute, PathAttribute, UpdateMessage,
+    LocalPrefPathAttribute, NextHopPathAttribute, OriginPathAttribute, PathAttribute,
+    UpdateMessage,
 };
 use bgp_primitives::ip_prefix::IpPrefix;
 use bgp_rib::{LocalRoute, LocalRouteAttrs};
@@ -452,3 +453,48 @@ async fn test_configured_routes_are_announced_to_every_new_session() -> Result<(
     }
     Ok(())
 }
+
+/// A peer in our own AS is sent a route we originate with an empty AS path
+/// and a LOCAL_PREF, whatever prepending is configured: our AS number goes
+/// on the path only as the route leaves the AS. It used to be sent as to
+/// an external peer, which an internal one refuses.
+#[tokio::test]
+async fn test_a_peer_in_our_own_as_gets_an_empty_path_and_a_local_pref() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    let mut peer = TestPeer::connect(server.addr).await?;
+    peer.handshake(LOCAL_ASN, Ipv4Addr::new(10, 0, 0, 1), 90)
+        .await?;
+    // The peer reads as a router in the server's AS does: one outside it
+    // would drop the LOCAL_PREF.
+    peer.ctx.ibgp = true;
+    wait_for_sessions(&server, 1).await?;
+    assert!(server.registry.list()[0].ibgp);
+
+    server
+        .local_rib
+        .announce(route(
+            "10.100.1.0/24",
+            LocalRouteAttrs {
+                prepend: 2,
+                ..LocalRouteAttrs::default()
+            },
+        ))
+        .await?;
+    let update = next_update(&mut peer).await?;
+    assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]);
+    assert!(
+        update
+            .path_attributes
+            .contains(&PathAttribute::AsPath(AsPathAttribute { segments: vec![] })),
+        "{:?}",
+        update.path_attributes
+    );
+    assert!(
+        update
+            .path_attributes
+            .contains(&PathAttribute::LocalPref(LocalPrefPathAttribute(100))),
+        "{:?}",
+        update.path_attributes
+    );
+    Ok(())
+}

crates/integration-tests/tests/dialer.rs

@@ -269,3 +269,51 @@ async fn test_local_address_is_used_as_the_source() -> Result<()> {
     assert!(reason.contains("local_address 192.0.2.1"), "{reason}");
     Ok(())
 }
+
+/// A peer that takes the connection and then refuses the session, here
+/// because it is not the AS we are configured to expect, will refuse it
+/// again. It is dialled ever more slowly, like a peer that cannot be
+/// reached: connecting used to count as success and reset the wait, so
+/// such a peer was dialled every few seconds for ever.
+#[tokio::test]
+async fn test_a_peer_that_refuses_the_session_is_dialled_ever_more_slowly() -> Result<()> {
+    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
+    let server = server_dialing(&listener, PEER_ASN).await?;
+
+    let mut waits = Vec::new();
+    for _ in 0..3 {
+        let mut peer = accept(&listener).await?;
+        peer.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90)
+            .await?;
+        let _ = peer.recv_open().await?;
+        let refusal = tokio::time::timeout(TIMEOUT, peer.recv()).await??;
+        assert!(
+            matches!(&refusal, Message::Notification(n) if n.error == BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs)),
+            "{refusal:?}"
+        );
+        drop(peer);
+
+        // How long the dialer says it will wait before trying again.
+        let wait = tokio::time::timeout(TIMEOUT, async {
+            loop {
+                if let Some((_, DialState::Backoff { until, last_error })) =
+                    server.registry.configured_peers().into_iter().next()
+                {
+                    assert!(last_error.contains("AS65099"), "{last_error}");
+                    return until
+                        .duration_since(std::time::SystemTime::now())
+                        .unwrap_or_default();
+                }
+                tokio::time::sleep(Duration::from_millis(5)).await;
+            }
+        })
+        .await?;
+        waits.push(wait);
+    }
+    // 200 ms, then 400, then 800, less what passed before each was read.
+    assert!(
+        waits[2] > Duration::from_millis(500),
+        "the waits were {waits:?}"
+    );
+    Ok(())
+}

crates/integration-tests/tests/interop.rs

@@ -119,7 +119,15 @@ async fn test_ipv6_prefix_update_shape() -> Result<()> {
         LocalRouteAttrs::default(),
     );
     let nh: std::net::Ipv6Addr = "fd00:179::2".parse()?;
-    let update = bgp_server::advertise::build_announce(&route, 4200000001, nh.into(), true)
+    let update = bgp_server::advertise::build_announce(
+        &route,
+        &bgp_server::advertise::Towards {
+            local_asn: 4200000001,
+            next_hop: nh.into(),
+            four_octet_asn: true,
+            ibgp: false,
+        },
+    )
     .expect("ipv6 over ipv6");
     assert!(update.announced_routes.is_empty());
     assert!(

crates/packet/src/open.rs

@@ -965,10 +965,7 @@ mod tests {
         assert_eq!(buf.to_vec(), out.to_vec());
         Ok(())
     }
-    /// Each capability must leave the bytes that follow it for the caller.
-    /// Returning the (empty) per-capability buffer instead makes the list
-    /// parser stop early and silently lose every later capability, e.g. the
-    /// 4-octet ASN that routers list after graceful restart.
+
     /// A peer advertising Multiple Labels used to hit a `todo!()` and take
     /// the session task down with it.
     #[test]
@@ -1228,6 +1225,10 @@ mod tests {
         assert!(too_long.to_wire(&mut BytesMut::new()).is_err());
     }
 
+    /// Each capability must leave the bytes that follow it for the caller.
+    /// Returning the (empty) per-capability buffer instead makes the list
+    /// parser stop early and silently lose every later capability, e.g. the
+    /// 4-octet ASN that routers list after graceful restart.
     #[test]
     fn capability_parsers_leave_the_following_bytes() {
         let trailer: &[u8] = &[0x41, 0x04, 0x00, 0x03, 0x34, 0x74];

crates/rib/src/lib.rs

@@ -3,9 +3,7 @@
 
 mod error;
 mod local_rib;
-mod peer;
 pub mod table;
 
 pub use error::RibStopped;
 pub use local_rib::{LocalRib, LocalRibEvent, LocalRoute, LocalRouteAttrs};
-pub use peer::PeerId;

crates/rib/src/peer.rs

@@ -1,14 +0,0 @@
-use std::net::SocketAddr;
-
-/// Identifies a BGP peer by its TCP address and negotiated ASN.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub struct PeerId {
-    pub addr: SocketAddr,
-    pub asn: u32,
-}
-
-impl std::fmt::Display for PeerId {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}(AS{})", self.addr, self.asn)
-    }
-}

crates/server/Cargo.toml

@@ -15,6 +15,7 @@ bytes.workspace = true
 clap.workspace = true
 futures.workspace = true
 prometheus.workspace = true
+socket2.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true

crates/server/src/advertise.rs

@@ -11,96 +11,127 @@ use std::net::IpAddr;
 
 use bgp_packet::message::{
     AsPathAttribute, AsPathSegment, CommunitiesPathAttribute, LargeCommunitiesPathAttribute,
-    Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, MultiExitDiscPathAttribute,
-    NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage,
+    LocalPrefPathAttribute, Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute,
+    MultiExitDiscPathAttribute, NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage,
 };
 use bgp_primitives::constants::{AfiSafi, IpVersion};
 use bgp_primitives::ip_prefix::{IpPrefix, Nlri};
+use bgp_rib::table::DEFAULT_LOCAL_PREF;
 use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute, RibStopped};
+use tokio::sync::broadcast::Receiver;
 use tokio::sync::broadcast::error::RecvError;
 use tokio::task::JoinHandle;
 use tracing::{debug, info, warn};
 
 use crate::session::{SessionEvent, SessionHandle, SessionRegistry};
 
+/// Starts the advertiser. It hears of every session that comes up after
+/// this returns, so call it before anything that can bring one up.
 pub fn spawn_advertiser(
     local_rib: LocalRib,
     registry: SessionRegistry,
     local_asn: u32,
 ) -> JoinHandle<()> {
-    tokio::spawn(run(local_rib, registry, local_asn))
+    let advertiser = Advertiser {
+        rib_events: local_rib.subscribe(),
+        session_events: registry.subscribe(),
+        local_rib,
+        registry,
+        local_asn,
+    };
+    tokio::spawn(advertiser.run())
 }
 
-/// Runs until the local RIB stops. The server watches for this task ending:
-/// without it nothing is announced.
-async fn run(local_rib: LocalRib, registry: SessionRegistry, local_asn: u32) {
-    let mut rib_events = local_rib.subscribe();
-    let mut session_events = registry.subscribe();
+struct Advertiser {
+    local_rib: LocalRib,
+    registry: SessionRegistry,
+    local_asn: u32,
+    rib_events: Receiver<LocalRibEvent>,
+    session_events: Receiver<SessionEvent>,
+}
 
+impl Advertiser {
+    /// Runs until the local RIB stops. The server watches for this task
+    /// ending: without it nothing is announced.
+    async fn run(mut self) {
         loop {
-        tokio::select! {
-            ev = rib_events.recv() => match ev {
-                Ok(LocalRibEvent::Announced(route)) => {
-                    for session in registry.list() {
-                        send_route(&session, &route, local_asn).await;
+            let handled = tokio::select! {
+                event = self.rib_events.recv() => self.on_route_event(event).await,
+                event = self.session_events.recv() => self.on_session_event(event).await,
+            };
+            if handled.is_err() {
+                return;
             }
         }
+    }
+
+    async fn on_route_event(
+        &self,
+        event: Result<LocalRibEvent, RecvError>,
+    ) -> Result<(), RibStopped> {
+        match event {
+            Ok(LocalRibEvent::Announced(route)) => {
+                for session in self.registry.list() {
+                    self.send_route(&session, &route).await;
+                }
+                Ok(())
+            }
             Ok(LocalRibEvent::Withdrawn { prefix }) => {
-                    for session in registry.list() {
+                for session in self.registry.list() {
                     send_withdraw(&session, &prefix).await;
                 }
+                Ok(())
             }
-                Err(RecvError::Lagged(n)) => {
-                    warn!(missed = n, "Advertiser lagged on local RIB events, resyncing all sessions");
-                    if resync_all(&local_rib, &registry, local_asn).await.is_err() {
-                        return;
+            Err(RecvError::Lagged(missed)) => {
+                warn!(
+                    missed,
+                    "Advertiser lagged on local RIB events, resyncing all sessions"
+                );
+                self.send_all_to(&self.registry.list()).await
+            }
+            Err(RecvError::Closed) => Err(RibStopped { rib: "local RIB" }),
         }
     }
-                Err(RecvError::Closed) => return,
-            },
-            ev = session_events.recv() => match ev {
+
+    async fn on_session_event(
+        &self,
+        event: Result<SessionEvent, RecvError>,
+    ) -> Result<(), RibStopped> {
+        match event {
             Ok(SessionEvent::Up(session)) => {
-                    let Ok(routes) = local_rib.snapshot().await else { return };
-                    info!(peer = %session.peer, routes = routes.len(), "Session up, sending local routes");
-                    for route in &routes {
-                        send_route(&session, route, local_asn).await;
-                    }
+                info!(peer = %session.peer, "Session up, sending local routes");
+                self.send_all_to(&[session]).await
             }
             Ok(SessionEvent::RefreshRequested(session)) => {
-                    let Ok(routes) = local_rib.snapshot().await else { return };
-                    info!(peer = %session.peer, routes = routes.len(), "Route refresh requested, re-sending local routes");
-                    for route in &routes {
-                        send_route(&session, route, local_asn).await;
+                info!(peer = %session.peer, "Route refresh requested, re-sending local routes");
+                self.send_all_to(&[session]).await
+            }
+            Ok(SessionEvent::Down(_)) => Ok(()),
+            Err(RecvError::Lagged(missed)) => {
+                warn!(
+                    missed,
+                    "Advertiser lagged on session events, resyncing all sessions"
+                );
+                self.send_all_to(&self.registry.list()).await
+            }
+            // The registry outlives the server's tasks; nothing more will
+            // come up, and nothing more needs sending.
+            Err(RecvError::Closed) => Err(RibStopped { rib: "local RIB" }),
         }
     }
-                Ok(SessionEvent::Down(_)) => {}
-                Err(RecvError::Lagged(n)) => {
-                    warn!(missed = n, "Advertiser lagged on session events, resyncing all sessions");
-                    if resync_all(&local_rib, &registry, local_asn).await.is_err() {
-                        return;
-                    }
-                }
-                Err(RecvError::Closed) => return,
-            },
-        }
-    }
-}
 
-async fn resync_all(
-    local_rib: &LocalRib,
-    registry: &SessionRegistry,
-    local_asn: u32,
-) -> Result<(), RibStopped> {
-    let routes = local_rib.snapshot().await?;
-    for session in registry.list() {
+    /// Sends every originated route to each of `sessions`.
+    async fn send_all_to(&self, sessions: &[SessionHandle]) -> Result<(), RibStopped> {
+        let routes = self.local_rib.snapshot().await?;
+        for session in sessions {
             for route in &routes {
-            send_route(&session, route, local_asn).await;
+                self.send_route(session, route).await;
             }
         }
         Ok(())
-}
+    }
 
-async fn send_route(session: &SessionHandle, route: &LocalRoute, local_asn: u32) {
+    async fn send_route(&self, session: &SessionHandle, route: &LocalRoute) {
         if !session.families.carries(&route.prefix) {
             debug!(
                 peer = %session.peer,
@@ -109,12 +140,13 @@ async fn send_route(session: &SessionHandle, route: &LocalRoute, local_asn: u32)
             );
             return;
         }
-    match build_announce(
-        route,
-        local_asn,
-        session.local_addr.ip(),
-        session.four_octet_asn,
-    ) {
+        let towards = Towards {
+            local_asn: self.local_asn,
+            next_hop: session.local_addr.ip(),
+            four_octet_asn: session.four_octet_asn,
+            ibgp: session.ibgp,
+        };
+        match build_announce(route, &towards) {
             Some(update) => {
                 debug!(peer = %session.peer, prefix = %route.prefix, "Announcing local route");
                 if let Err(e) = session.send(Message::Update(update)).await {
@@ -127,6 +159,7 @@ async fn send_route(session: &SessionHandle, route: &LocalRoute, local_asn: u32)
                 "Skipping route: address family does not match session transport"
             ),
         }
+    }
 }
 
 async fn send_withdraw(session: &SessionHandle, prefix: &IpPrefix) {
@@ -140,31 +173,44 @@ async fn send_withdraw(session: &SessionHandle, prefix: &IpPrefix) {
     }
 }
 
-/// Builds the UPDATE announcing `route` with `next_hop` as the next hop.
+/// What about a session decides how a route is announced on it.
+#[derive(Debug, Clone, Copy)]
+pub struct Towards {
+    pub local_asn: u32,
+    /// Our address on the session, which is the routes' next hop.
+    pub next_hop: IpAddr,
+    /// Whether the session negotiated four-octet AS numbers. If it did
+    /// not, and an AS number in the path needs four octets, the path is
+    /// also sent as AS4_PATH so that it survives the two-octet speaker
+    /// (RFC 6793 §4.2.2).
+    pub four_octet_asn: bool,
+    /// Whether the peer is in our own AS.
+    pub ibgp: bool,
+}
+
+/// Builds the UPDATE announcing `route` on a session.
 ///
 /// Returns `None` when the prefix family does not match the next-hop family:
 /// carrying IPv4 NLRI over an IPv6 next hop (RFC 8950) or vice versa is not
 /// supported yet.
-///
-/// `four_octet_asn` is whether the session the UPDATE is for negotiated
-/// four-octet AS numbers. If it did not, and the local AS number needs four
-/// octets, the path is also sent as AS4_PATH so that it survives the
-/// two-octet speaker (RFC 6793 §4.2.2).
-pub fn build_announce(
-    route: &LocalRoute,
-    local_asn: u32,
-    next_hop: IpAddr,
-    four_octet_asn: bool,
-) -> Option<UpdateMessage> {
+pub fn build_announce(route: &LocalRoute, towards: &Towards) -> Option<UpdateMessage> {
     let attrs = &route.attrs;
 
+    // Our AS number goes on the path as a route leaves the AS, and not
+    // before (RFC 4271 §5.1.2): a route we originate has an empty path
+    // inside it. A peer in our AS that found our AS number on the path
+    // would take the route for one that had looped.
     let as_path = AsPathAttribute {
-        segments: vec![AsPathSegment {
+        segments: if towards.ibgp {
+            vec![]
+        } else {
+            vec![AsPathSegment {
                 ordered: true,
-            path: vec![local_asn; 1 + usize::from(attrs.prepend)],
-        }],
+                path: vec![towards.local_asn; 1 + usize::from(attrs.prepend)],
+            }]
+        },
     };
-    let as4_path = if four_octet_asn {
+    let as4_path = if towards.four_octet_asn {
         None
     } else {
         as_path.as4_path()
@@ -176,34 +222,42 @@ pub fn build_announce(
         PathAttribute::Origin(attrs.origin),
         PathAttribute::AsPath(as_path),
     ];
-
+    // IPv4 routes go in the body of the UPDATE with their next hop in
+    // NEXT_HOP; IPv6 routes go in MP_REACH_NLRI, which carries its own.
     let mut announced_routes = vec![];
-    match (route.prefix.address_family(), next_hop) {
-        (IpVersion::Ipv4, IpAddr::V4(nh)) => {
-            path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(nh)));
+    let mp_reach = match (route.prefix.address_family(), towards.next_hop) {
+        (IpVersion::Ipv4, IpAddr::V4(next_hop)) => {
+            path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(next_hop)));
             announced_routes.push(route.prefix.into());
+            None
+        }
+        (IpVersion::Ipv6, IpAddr::V6(next_hop)) => {
+            Some(PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
+                family: AfiSafi::IPV6_UNICAST,
+                next_hop: NlriNextHop::Ipv6(next_hop),
+                prefixes: vec![route.prefix.into()],
+            }))
         }
-        (IpVersion::Ipv6, IpAddr::V6(_)) => {}
         _ => return None,
-    }
-
+    };
     if let Some(med) = attrs.med {
         path_attributes.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(
             med,
         )));
     }
+    // LOCAL_PREF is sent to peers in our AS and to no others (RFC 4271
+    // §5.1.5).
+    if towards.ibgp {
+        path_attributes.push(PathAttribute::LocalPref(LocalPrefPathAttribute(
+            DEFAULT_LOCAL_PREF,
+        )));
+    }
     if !attrs.communities.is_empty() {
         path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute(
             attrs.communities.clone(),
         )));
     }
-    if let (IpVersion::Ipv6, IpAddr::V6(nh)) = (route.prefix.address_family(), next_hop) {
-        path_attributes.push(PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
-            family: AfiSafi::IPV6_UNICAST,
-            next_hop: NlriNextHop::Ipv6(nh),
-            prefixes: vec![route.prefix.into()],
-        }));
-    }
+    path_attributes.extend(mp_reach);
     path_attributes.extend(as4_path);
     if !attrs.large_communities.is_empty() {
         path_attributes.push(PathAttribute::LargeCommunities(
@@ -254,6 +308,16 @@ mod tests {
         IpPrefix::try_from(s).unwrap()
     }
 
+    /// An external peer.
+    fn towards(local_asn: u32, next_hop: IpAddr, four_octet_asn: bool) -> Towards {
+        Towards {
+            local_asn,
+            next_hop,
+            four_octet_asn,
+            ibgp: false,
+        }
+    }
+
     #[test]
     fn ipv4_announce_has_expected_attributes_and_roundtrips() {
         let route = LocalRoute::new(
@@ -270,8 +334,11 @@ mod tests {
                 }],
             },
         );
-        let update =
-            build_announce(&route, 65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true).unwrap();
+        let update = build_announce(
+            &route,
+            &towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true),
+        )
+        .unwrap();
 
         assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]);
         assert!(update.withdrawn_routes.is_empty());
@@ -315,7 +382,7 @@ mod tests {
     fn ipv6_announce_uses_mp_reach() {
         let route = LocalRoute::new(prefix("2001:db8:1::/48"), LocalRouteAttrs::default());
         let nh = Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1);
-        let update = build_announce(&route, 65000, IpAddr::V6(nh), true).unwrap();
+        let update = build_announce(&route, &towards(65000, IpAddr::V6(nh), true)).unwrap();
 
         assert!(update.announced_routes.is_empty());
         assert!(
@@ -351,7 +418,7 @@ mod tests {
             })
         };
 
-        let update = build_announce(&route, 4_200_000_001, next_hop, false).unwrap();
+        let update = build_announce(&route, &towards(4_200_000_001, next_hop, false)).unwrap();
         assert_eq!(path(&update, true), Some(vec![4_200_000_001; 2]));
 
         // What the peer reads from the wire.
@@ -370,9 +437,9 @@ mod tests {
 
         // No AS4_PATH when it would say nothing new: the peer handles
         // four-octet numbers, or ours fits in two.
-        let update = build_announce(&route, 4_200_000_001, next_hop, true).unwrap();
+        let update = build_announce(&route, &towards(4_200_000_001, next_hop, true)).unwrap();
         assert_eq!(path(&update, true), None);
-        let update = build_announce(&route, 65000, next_hop, false).unwrap();
+        let update = build_announce(&route, &towards(65000, next_hop, false)).unwrap();
         assert_eq!(path(&update, true), None);
         assert_eq!(path(&update, false), Some(vec![65000; 2]));
     }
@@ -380,9 +447,13 @@ mod tests {
     #[test]
     fn family_mismatch_is_skipped() {
         let v6 = LocalRoute::new(prefix("2001:db8::/32"), LocalRouteAttrs::default());
-        assert!(build_announce(&v6, 65000, IpAddr::V4(Ipv4Addr::LOCALHOST), true).is_none());
+        assert!(
+            build_announce(&v6, &towards(65000, IpAddr::V4(Ipv4Addr::LOCALHOST), true)).is_none()
+        );
         let v4 = LocalRoute::new(prefix("10.0.0.0/8"), LocalRouteAttrs::default());
-        assert!(build_announce(&v4, 65000, IpAddr::V6(Ipv6Addr::LOCALHOST), true).is_none());
+        assert!(
+            build_announce(&v4, &towards(65000, IpAddr::V6(Ipv6Addr::LOCALHOST), true)).is_none()
+        );
     }
 
     #[test]
@@ -397,4 +468,69 @@ mod tests {
             matches!(&v6.path_attributes[..], [PathAttribute::MpUnreachNlri(m)] if m.prefixes == vec![prefix("2001:db8::/32")])
         );
     }
+
+    /// Inside our own AS a route we originate has an empty AS path and a
+    /// LOCAL_PREF. With our AS number on the path, a peer in our AS takes
+    /// the route for a loop; without LOCAL_PREF it treats the route as
+    /// withdrawn, as our own decoder does.
+    #[test]
+    fn a_route_announced_inside_our_as_has_an_empty_path_and_a_local_pref() {
+        let route = LocalRoute::new(
+            prefix("10.100.1.0/24"),
+            LocalRouteAttrs {
+                prepend: 3,
+                ..LocalRouteAttrs::default()
+            },
+        );
+        let internal = Towards {
+            ibgp: true,
+            ..towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true)
+        };
+        let update = build_announce(&route, &internal).unwrap();
+        assert_eq!(
+            update.path_attributes,
+            vec![
+                PathAttribute::Origin(OriginPathAttribute::IGP),
+                PathAttribute::AsPath(AsPathAttribute { segments: vec![] }),
+                PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
+                PathAttribute::LocalPref(LocalPrefPathAttribute(100)),
+            ]
+        );
+
+        // An internal peer accepts it as it stands.
+        let ctx = ParserContext {
+            four_octet_asn: true,
+            ibgp: true,
+            add_path: Vec::new(),
+        };
+        let mut buf = BytesMut::new();
+        Message::Update(update.clone())
+            .to_wire(&ctx, &mut buf)
+            .unwrap();
+        let decoded = Message::decode_with_report(&ctx, &buf).unwrap();
+        assert!(
+            decoded.update_errors.is_empty(),
+            "{:?}",
+            decoded.update_errors
+        );
+        assert_eq!(decoded.message, Message::Update(update));
+
+        // To an external peer the same route has our AS on the path, as
+        // many times as configured, and no LOCAL_PREF.
+        let external = Towards {
+            ibgp: false,
+            ..internal
+        };
+        let update = build_announce(&route, &external).unwrap();
+        assert_eq!(
+            update.path_attributes[1],
+            AsPathAttribute::from_asns(vec![65000; 4])
+        );
+        assert!(
+            !update
+                .path_attributes
+                .iter()
+                .any(|a| matches!(a, PathAttribute::LocalPref(_)))
+        );
+    }
 }

crates/server/src/collision.rs

@@ -8,9 +8,14 @@
 //! This cannot be a check local to one connection: the rival connection is
 //! usually still mid-handshake (so it is not in the session registry yet),
 //! and sometimes it is the rival, running in another task, that must close.
-//! Hence one small shared map, keyed by the peer's BGP Identifier, holding
-//! the single connection currently preferred for that peer and a way to
-//! tell it to stop.
+//! Hence one small shared map, holding the single connection currently
+//! preferred for each peer and a way to tell it to stop.
+//!
+//! A peer here is an address together with the BGP Identifier it announced.
+//! The identifier alone would not do: a router commonly has one session
+//! over IPv4 and another over IPv6 with the same neighbour, which are two
+//! sessions and not a collision; and it would let any host that can reach
+//! us displace a peer's connection by claiming the peer's identifier.
 //!
 //! A collision with an *Established* connection closes the new one, unless
 //! `detect_established` is set (the RFC's optional
@@ -19,7 +24,7 @@
 //! by the same side the newer wins: the initiator has abandoned the older.
 
 use std::collections::HashMap;
-use std::net::Ipv4Addr;
+use std::net::{IpAddr, Ipv4Addr};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 
@@ -76,10 +81,14 @@ fn new_loses(
     new != keep
 }
 
+/// Who a connection is with: the address at the other end, and the BGP
+/// Identifier announced from it.
+type Peer = (IpAddr, Ipv4Addr);
+
 /// Shared by every connection of one server.
 #[derive(Clone)]
 pub struct CollisionTable {
-    entries: Arc<Mutex<HashMap<Ipv4Addr, Entry>>>,
+    entries: Arc<Mutex<HashMap<Peer, Entry>>>,
     next_id: Arc<AtomicU64>,
     detect_established: bool,
 }
@@ -100,11 +109,13 @@ impl CollisionTable {
     pub fn admit(
         &self,
         local_id: Ipv4Addr,
+        remote_addr: IpAddr,
         remote_id: Ipv4Addr,
         direction: Direction,
     ) -> Option<ConnectionGuard> {
+        let peer = (remote_addr, remote_id);
         let mut entries = self.entries.locked();
-        if let Some(existing) = entries.get_mut(&remote_id) {
+        if let Some(existing) = entries.get_mut(&peer) {
             if new_loses(
                 local_id,
                 remote_id,
@@ -123,7 +134,7 @@ impl CollisionTable {
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
         let (kill, killed) = oneshot::channel();
         entries.insert(
-            remote_id,
+            peer,
             Entry {
                 id,
                 direction,
@@ -133,7 +144,7 @@ impl CollisionTable {
         );
         Some(ConnectionGuard {
             table: self.clone(),
-            remote_id,
+            peer,
             id,
             killed,
         })
@@ -143,20 +154,28 @@ impl CollisionTable {
 /// Holds a connection's slot in the table; releases it on drop.
 pub struct ConnectionGuard {
     table: CollisionTable,
-    remote_id: Ipv4Addr,
+    peer: Peer,
     id: u64,
-    /// Completes when a later connection won against this one; the owner
-    /// must then send Cease / Connection Collision Resolution and close.
-    pub killed: oneshot::Receiver<()>,
+    killed: oneshot::Receiver<()>,
 }
 
 impl ConnectionGuard {
     pub fn set_established(&self) {
         let mut entries = self.table.entries.locked();
-        if let Some(e) = entries.get_mut(&self.remote_id).filter(|e| e.id == self.id) {
+        if let Some(e) = entries.get_mut(&self.peer).filter(|e| e.id == self.id) {
             e.established = true;
         }
     }
+
+    /// Completes when a later connection has won against this one; the
+    /// owner must then send Cease / Connection Collision Resolution and
+    /// close. Cancel-safe.
+    pub async fn killed(&mut self) {
+        // The sending half is dropped only when it has been used: the entry
+        // that holds it is removed by this guard alone, or replaced by a
+        // connection that signals first. Either way, the wait is over.
+        let _ = (&mut self.killed).await;
+    }
 }
 
 impl Drop for ConnectionGuard {
@@ -164,11 +183,8 @@ impl Drop for ConnectionGuard {
         // A connection that lost has already been replaced in the map; only
         // remove the slot if it is still ours.
         let mut entries = self.table.entries.locked();
-        if entries
-            .get(&self.remote_id)
-            .is_some_and(|e| e.id == self.id)
-        {
-            entries.remove(&self.remote_id);
+        if entries.get(&self.peer).is_some_and(|e| e.id == self.id) {
+            entries.remove(&self.peer);
         }
     }
 }
@@ -247,24 +263,42 @@ mod tests {
     #[tokio::test]
     async fn table_signals_loser_and_releases_slots() {
         let table = CollisionTable::new(false);
+        let addr: IpAddr = Ipv4Addr::new(192, 0, 2, 1).into();
         // We are LOW: outbound first, then the peer's inbound wins.
-        let mut out = table.admit(LOW, HIGH, Outbound).unwrap();
-        let inb = table.admit(LOW, HIGH, Inbound).unwrap();
+        let mut out = table.admit(LOW, addr, HIGH, Outbound).unwrap();
+        let inb = table.admit(LOW, addr, HIGH, Inbound).unwrap();
         assert!(out.killed.try_recv().is_ok());
         // The loser's guard dropping must not free the winner's slot.
         drop(out);
         inb.set_established();
-        assert!(table.admit(LOW, HIGH, Outbound).is_none());
+        assert!(table.admit(LOW, addr, HIGH, Outbound).is_none());
 
         // A different identifier never collides.
         assert!(
             table
-                .admit(LOW, Ipv4Addr::new(192, 0, 2, 2), Inbound)
+                .admit(LOW, addr, Ipv4Addr::new(192, 0, 2, 2), Inbound)
                 .is_some()
         );
 
         // Once the survivor goes away the identifier is free again.
         drop(inb);
-        assert!(table.admit(LOW, HIGH, Outbound).is_some());
+        assert!(table.admit(LOW, addr, HIGH, Outbound).is_some());
+    }
+
+    /// A neighbour reached over IPv4 and over IPv6 announces the same
+    /// identifier on both sessions. They are two sessions, and neither
+    /// closes the other. Nor can a connection from elsewhere that claims a
+    /// peer's identifier take the place of the peer's own.
+    #[tokio::test]
+    async fn the_same_identifier_from_another_address_is_another_peer() {
+        let table = CollisionTable::new(true);
+        let v4: IpAddr = Ipv4Addr::new(192, 0, 2, 1).into();
+        let v6: IpAddr = "2001:db8::1".parse().unwrap();
+
+        let mut over_v4 = table.admit(LOW, v4, HIGH, Outbound).unwrap();
+        over_v4.set_established();
+        let over_v6 = table.admit(LOW, v6, HIGH, Inbound);
+        assert!(over_v6.is_some());
+        assert!(over_v4.killed.try_recv().is_err(), "the first is untouched");
     }
 }

crates/server/src/config.rs

@@ -56,9 +56,17 @@ pub struct PeerConfig {
 }
 
 impl Config {
+    /// The configured peer at `address`, if there is one.
+    pub fn peer_at(&self, address: IpAddr) -> Option<&PeerConfig> {
+        self.peers.iter().find(|peer| peer.endpoint.ip() == address)
+    }
+
     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)
-            .map_err(|_| ConfigError::RouterId(proto.router_id.clone()))?;
+            .ok()
+            .filter(|id| !id.is_unspecified())
+            .ok_or_else(|| ConfigError::RouterId(proto.router_id.clone()))?;
 
         if proto.local_asn == 0 {
             return Err(ConfigError::LocalAsn);
@@ -80,17 +88,27 @@ impl Config {
                 .collect::<Result<_, _>>()?
         };
 
-        let peers = proto
-            .peers
-            .into_iter()
-            .map(|peer| {
+        let mut peers: Vec<PeerConfig> = Vec::with_capacity(proto.peers.len());
+        for peer in proto.peers {
             let name = peer.name.clone();
-                parse_peer(peer).map_err(|problem| ConfigError::Peer {
-                    peer: name,
+            let rejected = |problem| ConfigError::Peer {
+                peer: name.clone(),
                 problem,
-                })
-            })
-            .collect::<Result<_, _>>()?;
+            };
+            let peer = parse_peer(peer).map_err(rejected)?;
+            // A peer is told from the others by its name in what the
+            // dialer reports, and by its address when it connects to us.
+            if peers.iter().any(|earlier| earlier.name == peer.name) {
+                return Err(rejected(PeerProblem::DuplicateName));
+            }
+            if peers
+                .iter()
+                .any(|earlier| earlier.endpoint.ip() == peer.endpoint.ip())
+            {
+                return Err(rejected(PeerProblem::DuplicateAddress));
+            }
+            peers.push(peer);
+        }
 
         let mut originate: Vec<LocalRoute> = Vec::with_capacity(proto.originate.len());
         for entry in &proto.originate {
@@ -185,7 +203,12 @@ fn parse_peer(peer: Peer) -> Result<PeerConfig, PeerProblem> {
     };
 
     Ok(PeerConfig {
-        name: peer.name,
+        // A peer without a name goes by its address.
+        name: if peer.name.is_empty() {
+            endpoint.ip().to_string()
+        } else {
+            peer.name
+        },
         endpoint,
         remote_asn: peer.remote_asn,
         hold_time_secs,
@@ -421,4 +444,54 @@ mod tests {
             ConfigError::HttpPort
         );
     }
+
+    /// Peers are told apart by name where the dialer reports on them and by
+    /// address when they connect to us; two that share either would be
+    /// taken for each other.
+    #[test]
+    fn peers_must_differ_in_name_and_address() {
+        let peer = |name: &str, address: &str| {
+            format!(
+                "peers {{ name: \"{name}\" endpoint {{ address: \"{address}\" }} remote_asn: 1 }}\n"
+            )
+        };
+        let problem = |peers: String| match rejected(&format!("{BASE}{peers}")) {
+            ConfigError::Peer { problem, .. } => problem,
+            other => panic!("{other}"),
+        };
+        assert_eq!(
+            problem(peer("r1", "192.0.2.2") + &peer("r1", "192.0.2.3")),
+            PeerProblem::DuplicateName
+        );
+        assert_eq!(
+            problem(peer("r1", "192.0.2.2") + &peer("r2", "192.0.2.2")),
+            PeerProblem::DuplicateAddress
+        );
+
+        // A peer without a name goes by its address, so two of them differ.
+        let config = parse(&format!(
+            "{BASE}{}{}",
+            peer("", "192.0.2.2"),
+            peer("", "192.0.2.3")
+        ))
+        .unwrap();
+        assert_eq!(config.peers[0].name, "192.0.2.2");
+        assert_eq!(config.peers[1].name, "192.0.2.3");
+        assert_eq!(
+            config
+                .peer_at("192.0.2.3".parse().unwrap())
+                .map(|p| &*p.name),
+            Some("192.0.2.3")
+        );
+        assert!(config.peer_at("192.0.2.4".parse().unwrap()).is_none());
+    }
+
+    /// A peer would refuse our OPEN.
+    #[test]
+    fn a_router_id_of_zero_is_rejected() {
+        assert_eq!(
+            rejected("router_id: \"0.0.0.0\" local_asn: 1"),
+            ConfigError::RouterId("0.0.0.0".into())
+        );
+    }
 }

crates/server/src/connection.rs

@@ -11,6 +11,7 @@ use std::io;
 use std::net::{Ipv4Addr, SocketAddr};
 use std::pin::Pin;
 use std::sync::Arc;
+use std::time::Duration;
 
 use bgp_packet::errors::{BgpError, DecodeError, MsgHeaderSubcode};
 use bgp_packet::message::{
@@ -19,78 +20,129 @@ use bgp_packet::message::{
 use bgp_packet::open::AddPathSendRecv;
 use bgp_packet::update_decode::UpdateError;
 use bgp_primitives::parser::ParserContext;
-use bgp_rib::PeerId;
-use bgp_rib::table::{Outcome, PeerIndex, PeerInfo, RibTable, TooManyPeers};
+use bgp_rib::table::{Outcome, PeerIndex, PeerInfo, RibTable};
 use bytes::BytesMut;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::sync::mpsc;
-use tokio::time::{Instant, Interval, Sleep, interval_at, sleep};
+use tokio::time::{Instant, Interval, MissedTickBehavior, Sleep, interval_at, sleep, timeout};
 use tracing::{debug, error, info, warn};
 
 use crate::collision::{CollisionTable, ConnectionGuard, Direction};
+use crate::config::PeerConfig;
 use crate::error::SessionError;
 use crate::family::FamilySet;
-use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated};
+use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated, Received};
 use crate::metrics::{
     MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, UPDATE_ERRORS, message_type_name,
 };
 use crate::panic::contained;
 use crate::routes;
-use crate::session::{SessionGuard, SessionHandle, SessionRegistry};
+use crate::session::{PeerId, SessionGuard, SessionHandle, SessionRegistry};
 
 /// Number of outbound messages that may be queued for a session before
 /// senders are back-pressured.
 const OUTBOUND_QUEUE_CAPACITY: usize = 1024;
 
-/// What a session needs from the rest of the server.
-#[derive(Clone)]
-pub struct SessionConfig {
+/// The least time a write to the peer may make no progress before the
+/// session is given up: the Send Hold Timer of RFC 9687, which is this or
+/// twice the hold time, whichever is longer. A peer that has stopped
+/// reading would otherwise hold the session, and our routes, for as long
+/// as TCP keeps trying.
+const MIN_SEND_HOLD_TIME: Duration = Duration::from_secs(8 * 60);
+
+/// How long a connection we are closing is kept to let the peer read our
+/// NOTIFICATION. Closing at once, with something of the peer's still
+/// unread, resets the connection, and the peer may then never see why.
+const CLOSE_LINGER: Duration = Duration::from_secs(1);
+
+/// What every session of a server shares.
+pub struct ServerContext {
     pub router_id: Ipv4Addr,
     pub local_asn: u32,
-    pub hold_time_secs: u16,
-    /// Where routes learned from the peer go.
+    /// Where routes learned from peers go.
     pub rib: Arc<RibTable>,
     pub registry: SessionRegistry,
-    /// Shared table used to resolve connection collisions (RFC 4271 §6.8).
+    /// Used to resolve connection collisions (RFC 4271 §6.8).
     pub collisions: CollisionTable,
-    /// Which side opened this TCP connection.
-    pub direction: Direction,
-    /// Configured name of the peer, if known.
-    pub peer_name: Option<String>,
+}
+
+/// What is configured for the peer at the other end of a connection.
+#[derive(Debug, Clone)]
+pub struct PeerSettings {
+    /// Configured name of the peer, if it is a configured one.
+    pub name: Option<String>,
     /// If set, the peer's OPEN must carry this ASN or the session is
     /// refused with a Bad Peer AS notification.
     pub expected_asn: Option<u32>,
+    pub hold_time_secs: u16,
     /// The address families to offer the peer.
     pub families: FamilySet,
     /// The directions to offer the peer ADD-PATH for (RFC 7911).
     pub add_path: Option<AddPathSendRecv>,
 }
 
-/// Runs a session on an accepted connection and logs how it ended.
-pub async fn handle_connection(
-    stream: TcpStream,
-    peer_addr: SocketAddr,
-    config: Arc<SessionConfig>,
-) {
-    if let Err(e) = run_session(stream, peer_addr, (*config).clone()).await {
-        warn!(%peer_addr, err = %e, "BGP session terminated with error");
+impl PeerSettings {
+    /// For a peer that connected to us and is not in the configuration.
+    pub fn unconfigured(hold_time_secs: u16) -> Self {
+        Self {
+            name: None,
+            expected_asn: None,
+            hold_time_secs,
+            families: FamilySet::supported(),
+            add_path: None,
+        }
     }
 }
 
-/// Runs the BGP session on `stream` until it ends.
+impl From<&PeerConfig> for PeerSettings {
+    fn from(peer: &PeerConfig) -> Self {
+        Self {
+            name: Some(peer.name.clone()),
+            expected_asn: Some(peer.remote_asn),
+            hold_time_secs: peer.hold_time_secs,
+            families: peer.families.clone(),
+            add_path: peer.add_path,
+        }
+    }
+}
+
+/// How a session ended.
+#[derive(Debug)]
+pub struct SessionEnd {
+    /// Whether it had reached Established.
+    pub established: bool,
+    /// `Ok` if it was closed at our own request (a shutdown); otherwise
+    /// what went wrong, or what the peer said.
+    pub result: Result<(), SessionError>,
+}
+
+/// Runs the BGP session on `stream` until it ends. `direction` is which
+/// side opened the connection.
 pub async fn run_session(
     stream: TcpStream,
     peer_addr: SocketAddr,
-    config: SessionConfig,
-) -> Result<(), SessionError> {
-    let (mut session, first_actions) = Session::open(stream, peer_addr, config)?;
+    direction: Direction,
+    server: Arc<ServerContext>,
+    peer: PeerSettings,
+) -> SessionEnd {
+    let (mut session, first_actions) =
+        match Session::open(stream, peer_addr, direction, server, peer) {
+            Ok(opened) => opened,
+            Err(e) => {
+                return SessionEnd {
+                    established: false,
+                    result: Err(e.into()),
+                };
+            }
+        };
 
     let result = match contained(session.run(first_actions)).await {
         Ok(result) => result,
         Err(panic) => Err(session.close_after_panic(panic).await),
     };
+    let established = session.established.is_some();
 
     // Whatever ended the session, routes learned over it are gone
     // (RFC 4271 §8.2.2). Do this before the registration drops so that the
@@ -110,43 +162,10 @@ pub async fn run_session(
             "Session ended"
         );
     }
-    result
-}
-
-impl Session<OwnedReadHalf> {
-    /// A session on a connection that has just been opened, and the first
-    /// things to do on it.
-    fn open(
-        stream: TcpStream,
-        peer_addr: SocketAddr,
-        config: SessionConfig,
-    ) -> io::Result<(Self, Vec<Action>)> {
-        let local_addr = stream.local_addr()?;
-        let (reader, writer) = stream.into_split();
-        let (fsm, first_actions) = Fsm::new(FsmConfig {
-            router_id: config.router_id,
-            local_asn: config.local_asn,
-            hold_time_secs: config.hold_time_secs,
-            expected_asn: config.expected_asn,
-            families: config.families.clone(),
-            add_path: config.add_path,
-        });
-        let session = Session {
-            fsm,
-            frames: FrameReader::new(reader),
-            writer,
-            ctx: ParserContext::default(),
-            send_ctx: ParserContext::default(),
-            hold_timer: None,
-            keepalive_timer: None,
-            collision: None,
-            established: None,
-            attributes_wire: None,
-            peer_addr,
-            local_addr,
-            config,
-        };
-        Ok((session, first_actions))
+    session.close().await;
+    SessionEnd {
+        established,
+        result,
     }
 }
 
@@ -231,10 +250,12 @@ impl Drop for PeerRoutes {
     }
 }
 
-struct Session<R> {
+struct Session {
     fsm: Fsm,
-    frames: FrameReader<R>,
+    frames: FrameReader<OwnedReadHalf>,
     writer: OwnedWriteHalf,
+    /// Reused for every message written.
+    write_buf: BytesMut,
     /// How messages from the peer are decoded: the width of AS numbers and
     /// where prefixes carry path identifiers are known once the OPENs are
     /// exchanged.
@@ -245,19 +266,63 @@ struct Session<R> {
     send_ctx: ParserContext,
     hold_timer: Option<Pin<Box<Sleep>>>,
     keepalive_timer: Option<Interval>,
+    /// How long a write may make no progress; see [`MIN_SEND_HOLD_TIME`].
+    send_hold_time: Duration,
+    /// Whether a NOTIFICATION of ours is on its way to the peer.
+    notified: bool,
     /// This connection's slot in the collision table, once admitted.
     collision: Option<ConnectionGuard>,
     established: Option<Established>,
-    /// The attributes of the UPDATE that is being acted on, as they were
-    /// on the wire, if the decoder kept them. Set when the message is read
-    /// and taken when its routes are stored.
-    attributes_wire: Option<Vec<u8>>,
     peer_addr: SocketAddr,
     local_addr: SocketAddr,
-    config: SessionConfig,
+    direction: Direction,
+    server: Arc<ServerContext>,
+    /// Configured name of the peer, if it is a configured one.
+    peer_name: Option<String>,
 }
 
-impl<R: AsyncRead + Unpin> Session<R> {
+impl Session {
+    /// A session on a connection that has just been opened, and the first
+    /// things to do on it.
+    fn open(
+        stream: TcpStream,
+        peer_addr: SocketAddr,
+        direction: Direction,
+        server: Arc<ServerContext>,
+        peer: PeerSettings,
+    ) -> io::Result<(Self, Vec<Action>)> {
+        let local_addr = stream.local_addr()?;
+        let (reader, writer) = stream.into_split();
+        let (fsm, first_actions) = Fsm::new(FsmConfig {
+            router_id: server.router_id,
+            local_asn: server.local_asn,
+            hold_time_secs: peer.hold_time_secs,
+            expected_asn: peer.expected_asn,
+            families: peer.families,
+            add_path: peer.add_path,
+        });
+        let session = Session {
+            fsm,
+            frames: FrameReader::new(reader),
+            writer,
+            write_buf: BytesMut::new(),
+            ctx: ParserContext::default(),
+            send_ctx: ParserContext::default(),
+            hold_timer: None,
+            keepalive_timer: None,
+            send_hold_time: MIN_SEND_HOLD_TIME,
+            notified: false,
+            collision: None,
+            established: None,
+            peer_addr,
+            local_addr,
+            direction,
+            server,
+            peer_name: peer.name,
+        };
+        Ok((session, first_actions))
+    }
+
     /// Says goodbye to the peer after the session's own code panicked.
     ///
     /// The state machine may be half way through something, so it is not
@@ -277,6 +342,20 @@ impl<R: AsyncRead + Unpin> Session<R> {
         error
     }
 
+    /// Closes the connection. If the peer was sent a NOTIFICATION, it is
+    /// given a moment to read it.
+    async fn close(mut self) {
+        // A new connection from the peer must not find this one in its way.
+        self.collision = None;
+        if !self.notified {
+            return;
+        }
+        // Our side is done sending; what the peer still sends is read and
+        // thrown away until it closes its side too.
+        let _ = self.writer.shutdown().await;
+        let _ = timeout(CLOSE_LINGER, self.frames.discard_until_closed()).await;
+    }
+
     async fn run(&mut self, first_actions: Vec<Action>) -> Result<(), SessionError> {
         let mut actions = first_actions;
         loop {
@@ -295,8 +374,10 @@ impl<R: AsyncRead + Unpin> Session<R> {
             frame = self.frames.next(&self.ctx) => match frame {
                 Ok(decoded) => {
                     self.report(&decoded.update_errors);
-                    self.attributes_wire = decoded.attributes_wire;
-                    Ok(Event::Message(decoded.message))
+                    Ok(Event::Message(Received {
+                        message: decoded.message,
+                        attributes_wire: decoded.attributes_wire,
+                    }))
                 }
                 Err(SessionError::Decode(e)) => Ok(Event::DecodeFailed(e)),
                 Err(e) => Err(e),
@@ -332,17 +413,26 @@ impl<R: AsyncRead + Unpin> Session<R> {
         let mut queue: VecDeque<Action> = actions.into();
         while let Some(action) = queue.pop_front() {
             match action {
-                Action::Send(msg) => {
-                    let closing = matches!(msg, Message::Notification(_));
-                    if let Err(e) = self.send(&msg).await {
-                        if !closing {
-                            return Some(Err(e));
-                        }
+                Action::Send(msg) => match self.send(&msg).await {
+                    Ok(()) => {}
                     // The session is ending anyway and the Close that
                     // follows says why; a NOTIFICATION is best effort.
+                    Err(e) if matches!(msg, Message::Notification(_)) => {
                         warn!(peer_addr = %self.peer_addr, err = %e, "Failed to send NOTIFICATION");
                     }
+                    // A message of our own making that cannot be written.
+                    // That is a bug here, and nothing the peer did: the
+                    // session is worth more than the message.
+                    Err(SessionError::Encode(e)) => {
+                        error!(
+                            peer_addr = %self.peer_addr,
+                            message = message_type_name(&msg),
+                            err = %e,
+                            "Dropped a message that cannot be encoded"
+                        );
                     }
+                    Err(e) => return Some(Err(e)),
+                },
                 Action::ResetHoldTimer(duration) => match &mut self.hold_timer {
                     Some(timer) => timer.as_mut().reset(Instant::now() + duration),
                     None => self.hold_timer = Some(Box::pin(sleep(duration))),
@@ -351,50 +441,23 @@ impl<R: AsyncRead + Unpin> Session<R> {
                 Action::StartKeepaliveTimer(every) => {
                     // interval_at, so that the first tick is one interval
                     // from now and not immediate.
-                    self.keepalive_timer = Some(interval_at(Instant::now() + every, every));
+                    let mut timer = interval_at(Instant::now() + every, every);
+                    // After a stall, one KEEPALIVE and not one for every
+                    // tick that was missed.
+                    timer.set_missed_tick_behavior(MissedTickBehavior::Delay);
+                    self.keepalive_timer = Some(timer);
                 }
                 Action::Admit(negotiated) => {
-                    self.ctx = ParserContext {
-                        four_octet_asn: negotiated.four_octet_asn,
-                        ibgp: negotiated.peer_asn == self.config.local_asn,
-                        add_path: negotiated.add_path_receive.iter().collect(),
-                    };
-                    self.send_ctx = ParserContext {
-                        add_path: negotiated.add_path_send.iter().collect(),
-                        ..self.ctx.clone()
-                    };
-                    match self.config.collisions.admit(
-                        self.config.router_id,
-                        negotiated.peer_identifier,
-                        self.config.direction,
-                    ) {
-                        Some(guard) => {
-                            self.collision = Some(guard);
-                            info!(
-                                peer_addr = %self.peer_addr,
-                                peer_asn = negotiated.peer_asn,
-                                negotiated_hold_time = negotiated.hold_time_secs,
-                                four_octet_asn = negotiated.four_octet_asn,
-                                families = %negotiated.families,
-                                add_path_receive = %negotiated.add_path_receive,
-                                add_path_send = %negotiated.add_path_send,
-                                direction = self.config.direction.as_str(),
-                                "OPEN received, transitioning to OpenConfirm"
-                            );
-                        }
-                        None => {
-                            // Not admitted: what the machine wanted next no
-                            // longer applies.
-                            info!(peer_addr = %self.peer_addr, "Closing connection: lost collision resolution");
+                    if !self.admit(&negotiated) {
+                        // What the machine wanted next no longer applies.
                         queue = self.fsm.handle(Event::CollisionLost).into();
                     }
                 }
-                }
                 Action::Established(negotiated) => {
-                    if let Err(full) = self.establish(negotiated) {
+                    if let Err(e) = self.establish(&negotiated) {
                         // What the machine wanted next no longer applies.
-                        error!(peer_addr = %self.peer_addr, err = %full, "Cannot take the peer's routes");
-                        queue = self.fsm.handle(Event::TableFull(full)).into();
+                        error!(peer_addr = %self.peer_addr, err = %e, "Cannot take the peer's routes");
+                        queue = self.fsm.handle(Event::LocalFailure(e)).into();
                     }
                 }
                 Action::RoutesIgnored(count) => {
@@ -404,34 +467,19 @@ impl<R: AsyncRead + Unpin> Session<R> {
                         "Ignored routes of an address family that was not negotiated"
                     );
                 }
-                Action::DeliverUpdate(update) => {
-                    let attributes_wire = self.attributes_wire.take();
-                    if let Some(established) = &self.established {
-                        let applied = routes::apply(
-                            &self.config.rib,
-                            established.rib_peer,
-                            &update,
-                            attributes_wire.as_deref(),
-                        );
-                        match applied {
-                            // One line for every UPDATE is too much to log
-                            // by default: a full table is a million of them.
-                            Ok(outcome) => debug!(
-                                peer = %established.peer_id,
-                                added = outcome.added,
-                                replaced = outcome.replaced,
-                                removed = outcome.removed,
-                                best_changed = outcome.best_changed,
-                                "Received UPDATE"
-                            ),
-                            Err(e) => return Some(Err(e.into())),
-                        }
+                Action::DeliverUpdate {
+                    update,
+                    attributes_wire,
+                } => {
+                    if let Err(e) = self.store(&update, attributes_wire.as_deref()) {
+                        error!(peer_addr = %self.peer_addr, err = %e, "Cannot store the peer's routes");
+                        queue = self.fsm.handle(Event::LocalFailure(e)).into();
                     }
                 }
                 Action::RouteRefreshRequested => {
                     if let Some(established) = &self.established {
                         info!(peer = %established.peer_id, "Received ROUTE-REFRESH, re-sending our routes");
-                        self.config.registry.request_refresh(&established.peer_id);
+                        self.server.registry.request_refresh(&established.peer_id);
                     }
                 }
                 Action::Close(result) => return Some(result),
@@ -440,18 +488,57 @@ impl<R: AsyncRead + Unpin> Session<R> {
         None
     }
 
+    /// Takes what the OPENs settled into use, and asks collision detection
+    /// whether this connection may go on. Returns whether it may.
+    fn admit(&mut self, negotiated: &Negotiated) -> bool {
+        self.ctx = ParserContext {
+            four_octet_asn: negotiated.four_octet_asn,
+            ibgp: negotiated.ibgp,
+            add_path: negotiated.add_path_receive.iter().collect(),
+        };
+        self.send_ctx = ParserContext {
+            add_path: negotiated.add_path_send.iter().collect(),
+            ..self.ctx.clone()
+        };
+        let hold_time = Duration::from_secs(negotiated.hold_time_secs.into());
+        self.send_hold_time = MIN_SEND_HOLD_TIME.max(2 * hold_time);
+
+        self.collision = self.server.collisions.admit(
+            self.server.router_id,
+            self.peer_addr.ip(),
+            negotiated.peer_identifier,
+            self.direction,
+        );
+        if self.collision.is_none() {
+            info!(peer_addr = %self.peer_addr, "Closing connection: lost collision resolution");
+            return false;
+        }
+        info!(
+            peer_addr = %self.peer_addr,
+            peer_asn = negotiated.peer_asn,
+            negotiated_hold_time = negotiated.hold_time_secs,
+            four_octet_asn = negotiated.four_octet_asn,
+            families = %negotiated.families,
+            add_path_receive = %negotiated.add_path_receive,
+            add_path_send = %negotiated.add_path_send,
+            direction = self.direction.as_str(),
+            "OPEN received, transitioning to OpenConfirm"
+        );
+        true
+    }
+
     /// Makes room for the peer's routes and makes the session visible to
     /// the rest of the server.
-    fn establish(&mut self, negotiated: Negotiated) -> Result<(), TooManyPeers> {
+    fn establish(&mut self, negotiated: &Negotiated) -> Result<(), SessionError> {
         let peer_id = PeerId {
             addr: self.peer_addr,
             asn: negotiated.peer_asn,
         };
-        let rib_peer = self.config.rib.add_peer(PeerInfo {
+        let rib_peer = self.server.rib.add_peer(PeerInfo {
             asn: negotiated.peer_asn,
             router_id: negotiated.peer_identifier,
             address: self.peer_addr.ip(),
-            ibgp: negotiated.peer_asn == self.config.local_asn,
+            ibgp: negotiated.ibgp,
         })?;
         info!(peer_addr = %self.peer_addr, peer_asn = negotiated.peer_asn, "Session established");
         SESSIONS_ESTABLISHED_TOTAL.inc();
@@ -460,12 +547,12 @@ impl<R: AsyncRead + Unpin> Session<R> {
             collision.set_established();
         }
         let (outbound_tx, outbound) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY);
-        let registration = self.config.registry.register(SessionHandle::new(
+        let registration = self.server.registry.register(SessionHandle::new(
             peer_id,
-            self.config.peer_name.clone(),
-            self.config.direction,
+            self.peer_name.clone(),
+            self.direction,
             self.local_addr,
-            &negotiated,
+            negotiated,
             rib_peer,
             outbound_tx,
         ));
@@ -473,12 +560,40 @@ impl<R: AsyncRead + Unpin> Session<R> {
             peer_id,
             rib_peer,
             outbound,
-            routes: PeerRoutes::new(Arc::clone(&self.config.rib), rib_peer),
+            routes: PeerRoutes::new(Arc::clone(&self.server.rib), rib_peer),
             _registration: registration,
         });
         Ok(())
     }
 
+    /// Puts the routes of a received UPDATE into the routing table.
+    fn store(
+        &self,
+        update: &UpdateMessage,
+        attributes_wire: Option<&[u8]>,
+    ) -> Result<(), SessionError> {
+        let Some(established) = &self.established else {
+            return Ok(());
+        };
+        let outcome = routes::apply(
+            &self.server.rib,
+            established.rib_peer,
+            update,
+            attributes_wire,
+        )?;
+        // One line for every UPDATE is too much to log by default: a full
+        // table is a million of them.
+        debug!(
+            peer = %established.peer_id,
+            added = outcome.added,
+            replaced = outcome.replaced,
+            removed = outcome.removed,
+            best_changed = outcome.best_changed,
+            "Received UPDATE"
+        );
+        Ok(())
+    }
+
     async fn send(&mut self, msg: &Message) -> Result<(), SessionError> {
         if let (Message::Update(update), Some(established)) = (msg, &self.established) {
             let (withdrawn, announced) = prefix_counts(update);
@@ -486,10 +601,14 @@ impl<R: AsyncRead + Unpin> Session<R> {
         }
         if let Message::Notification(notification) = msg {
             info!(peer_addr = %self.peer_addr, error = %notification.error, "Sending NOTIFICATION, closing session");
+            self.notified = true;
+        }
+        self.write_buf.clear();
+        msg.to_wire(&self.send_ctx, &mut self.write_buf)?;
+        match timeout(self.send_hold_time, self.writer.write_all(&self.write_buf)).await {
+            Ok(written) => written?,
+            Err(_) => return Err(SessionError::SendHoldTimerExpired(self.send_hold_time)),
         }
-        let mut buf = BytesMut::new();
-        msg.to_wire(&self.send_ctx, &mut buf)?;
-        self.writer.write_all(&buf).await?;
         MESSAGES_SENT
             .with_label_values(&[message_type_name(msg)])
             .inc();
@@ -533,11 +652,7 @@ async fn queued(established: &mut Option<Established>) -> Message {
 
 async fn killed(collision: &mut Option<ConnectionGuard>) {
     match collision {
-        Some(guard) => {
-            // An error means the collision table is gone, i.e. the server is
-            // going away; ending the session is right then too.
-            let _ = (&mut guard.killed).await;
-        }
+        Some(guard) => guard.killed().await,
         None => pending().await,
     }
 }
@@ -592,6 +707,17 @@ impl<R: AsyncRead + Unpin> FrameReader<R> {
             }
         }
     }
+
+    /// Reads and throws away whatever the peer sends, until it closes the
+    /// connection or the connection fails.
+    async fn discard_until_closed(&mut self) {
+        loop {
+            self.buf.clear();
+            if !matches!(self.reader.read_buf(&mut self.buf).await, Ok(read) if read > 0) {
+                return;
+            }
+        }
+    }
 }
 
 /// The length of the message at the front of `buf`, once its header has
@@ -721,20 +847,21 @@ mod tests {
             .await
             .unwrap();
         let (stream, peer_addr) = listener.accept().await.unwrap();
-        let config = SessionConfig {
+        let server = Arc::new(ServerContext {
             router_id: Ipv4Addr::new(10, 0, 0, 1),
             local_asn: 65000,
-            hold_time_secs: 90,
             rib: Arc::new(RibTable::new(4)),
             registry: SessionRegistry::new(),
             collisions: CollisionTable::new(false),
-            direction: Direction::Inbound,
-            peer_name: None,
-            expected_asn: None,
-            families: FamilySet::supported(),
-            add_path: None,
-        };
-        let (mut session, _) = Session::open(stream, peer_addr, config).unwrap();
+        });
+        let (mut session, _) = Session::open(
+            stream,
+            peer_addr,
+            Direction::Inbound,
+            server,
+            PeerSettings::unconfigured(90),
+        )
+        .unwrap();
 
         let error = session
             .close_after_panic("index out of bounds".into())
@@ -753,6 +880,56 @@ mod tests {
         assert_eq!(received, expected);
     }
 
+    /// A peer that stops reading must not hold the session for as long as
+    /// TCP will keep trying: a write that makes no progress is given up
+    /// (RFC 9687). Nothing else would end it, since the session reads
+    /// nothing, timers included, while it waits to write.
+    #[tokio::test]
+    async fn a_peer_that_stops_reading_is_given_up_on() {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        // Held, and never read from.
+        let _peer = TcpStream::connect(listener.local_addr().unwrap())
+            .await
+            .unwrap();
+        let (stream, peer_addr) = listener.accept().await.unwrap();
+        let server = Arc::new(ServerContext {
+            router_id: Ipv4Addr::new(10, 0, 0, 1),
+            local_asn: 65000,
+            rib: Arc::new(RibTable::new(4)),
+            registry: SessionRegistry::new(),
+            collisions: CollisionTable::new(false),
+        });
+        let (mut session, _) = Session::open(
+            stream,
+            peer_addr,
+            Direction::Inbound,
+            server,
+            PeerSettings::unconfigured(90),
+        )
+        .unwrap();
+        session.send_hold_time = Duration::from_millis(100);
+
+        // Until the socket's buffers are full, writes succeed.
+        let filler = Message::Notification(NotificationMessage {
+            error: BgpError::Cease(CeaseSubcode::AdminShutdown),
+            data: vec![0; 4000],
+        });
+        let stuck = tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                if let Err(e) = session.send(&filler).await {
+                    return e;
+                }
+            }
+        })
+        .await
+        .expect("a write that cannot finish is given up");
+        assert!(
+            matches!(stuck, SessionError::SendHoldTimerExpired(_)),
+            "{stuck}"
+        );
+        assert!(stuck.notification().is_none(), "there is no way to say so");
+    }
+
     /// A session task that is torn down without running its cleanup must
     /// not leave the peer's routes behind.
     #[tokio::test]

crates/server/src/dialer.rs

@@ -17,8 +17,8 @@ use tracing::{info, warn};
 
 use crate::collision::Direction;
 use crate::config::PeerConfig;
-use crate::connection::{SessionConfig, run_session};
-use crate::session::{DialState, SessionRegistry};
+use crate::connection::{PeerSettings, ServerContext, run_session};
+use crate::session::DialState;
 
 /// Upper bound for exponential backoff between connection attempts.
 const MAX_BACKOFF: Duration = Duration::from_secs(120);
@@ -26,23 +26,12 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
 
 pub fn spawn_dialers(
     peers: &[PeerConfig],
-    base: &SessionConfig,
+    server: &Arc<ServerContext>,
     connect_retry: Duration,
 ) -> Vec<JoinHandle<()>> {
     peers
         .iter()
-        .map(|peer| {
-            let config = SessionConfig {
-                hold_time_secs: peer.hold_time_secs,
-                direction: Direction::Outbound,
-                peer_name: Some(peer.name.clone()),
-                expected_asn: Some(peer.remote_asn),
-                families: peer.families.clone(),
-                add_path: peer.add_path,
-                ..base.clone()
-            };
-            tokio::spawn(dial_loop(Arc::new(peer.clone()), config, connect_retry))
-        })
+        .map(|peer| tokio::spawn(dial_loop(peer.clone(), Arc::clone(server), connect_retry)))
         .collect()
 }
 
@@ -64,8 +53,8 @@ async fn connect(peer: &PeerConfig) -> std::io::Result<TcpStream> {
     socket.connect(peer.endpoint).await
 }
 
-async fn dial_loop(peer: Arc<PeerConfig>, config: SessionConfig, connect_retry: Duration) {
-    let registry: SessionRegistry = config.registry.clone();
+async fn dial_loop(peer: PeerConfig, server: Arc<ServerContext>, connect_retry: Duration) {
+    let registry = &server.registry;
     let mut backoff = connect_retry;
 
     loop {
@@ -82,21 +71,32 @@ async fn dial_loop(peer: Arc<PeerConfig>, config: SessionConfig, connect_retry:
         registry.set_dial_state(&peer.name, DialState::Connecting);
         info!(peer = %peer.name, endpoint = %peer.endpoint, "Connecting to peer");
 
-        let outcome = match tokio::time::timeout(CONNECT_TIMEOUT, connect(&peer)).await {
+        let ended = match tokio::time::timeout(CONNECT_TIMEOUT, connect(&peer)).await {
             Ok(Ok(stream)) => {
                 info!(peer = %peer.name, endpoint = %peer.endpoint, "TCP connected, starting session");
+                let end = run_session(
+                    stream,
+                    peer.endpoint,
+                    Direction::Outbound,
+                    Arc::clone(&server),
+                    PeerSettings::from(&peer),
+                )
+                .await;
+                // A session that came up is a peer worth dialling again
+                // soon. One that connects and is then refused, for its AS
+                // number say, will be refused again: it is retried ever
+                // more slowly, like one that cannot be reached at all.
+                if end.established {
                     backoff = connect_retry;
-                run_session(stream, peer.endpoint, config.clone())
-                    .await
-                    .map(|()| "session closed by peer".to_string())
-                    .map_err(|e| e.to_string())
+                }
+                end.result.map_err(|e| e.to_string())
             }
             Ok(Err(e)) => Err(format!("connect failed: {e}")),
             Err(_) => Err(format!("connect timed out after {CONNECT_TIMEOUT:?}")),
         };
 
-        let reason = match outcome {
-            Ok(msg) => msg,
+        let reason = match ended {
+            Ok(()) => "session closed at our request".to_string(),
             Err(reason) => {
                 warn!(peer = %peer.name, endpoint = %peer.endpoint, err = %reason, "Session attempt ended");
                 reason

crates/server/src/error.rs

@@ -2,6 +2,7 @@
 
 use std::io;
 use std::net::SocketAddr;
+use std::time::Duration;
 
 use bgp_packet::errors::{BgpError, CeaseSubcode, DecodeError, FsmSubcode, OpenMsgSubcode};
 use bgp_packet::message::NotificationMessage;
@@ -41,8 +42,6 @@ pub enum SessionError {
     WrongPeerAs { expected: u32, actual: u32 },
     #[error("closed by connection collision resolution")]
     Collision,
-    /// The peer offers none of the address families configured for it, so
-    /// the session could carry nothing.
     /// The routing table has no room for another peer.
     #[error(transparent)]
     TableFull(#[from] TooManyPeers),
@@ -50,8 +49,14 @@ pub enum SessionError {
     /// is being reset, and the session is cleaned up like any other.
     #[error("the session panicked: {0}")]
     Panicked(String),
+    /// The peer offers none of the address families configured for it, so
+    /// the session could carry nothing.
     #[error("no address family in common: we offer {ours}, the peer offers {theirs}")]
     NoCommonAddressFamily { ours: FamilySet, theirs: FamilySet },
+    /// The peer took nothing we sent for this long (RFC 9687): it has
+    /// stopped reading, and there is no telling it so.
+    #[error("the peer accepted no data for {0:?}")]
+    SendHoldTimerExpired(Duration),
 }
 
 impl SessionError {
@@ -59,7 +64,10 @@ impl SessionError {
     /// or `None` when there is nothing to say or no way to say it.
     pub fn notification(&self) -> Option<NotificationMessage> {
         let error = match self {
-            Self::Io(_) | Self::Encode(_) | Self::NotifiedByPeer(_) => return None,
+            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),
             Self::Decode(e) => {
                 return Some(NotificationMessage {
                     error: e.notification.clone(),
@@ -97,7 +105,7 @@ impl SessionError {
     }
 }
 
-/// Why the server could not start.
+/// Why the server could not start, or could not go on.
 #[derive(Debug, Error)]
 pub enum ServerError {
     #[error("failed to bind listener on {addr}: {source}")]
@@ -120,7 +128,7 @@ pub enum ServerError {
 /// Why a configuration was rejected.
 #[derive(Debug, Clone, PartialEq, Eq, Error)]
 pub enum ConfigError {
-    #[error("invalid router_id {0:?}: must be a dotted-quad IPv4 address")]
+    #[error("invalid router_id {0:?}: must be a dotted-quad IPv4 address other than 0.0.0.0")]
     RouterId(String),
     #[error("local_asn must be non-zero")]
     LocalAsn,
@@ -150,6 +158,10 @@ pub enum PeerProblem {
     RemoteAsn,
     #[error("hold_time_seconds must be 0 or at least 3")]
     HoldTime,
+    #[error("there is another peer of this name")]
+    DuplicateName,
+    #[error("there is another peer at this address")]
+    DuplicateAddress,
     #[error("local_address {0:?} is not an IP address")]
     LocalAddress(String),
     #[error("local_address and endpoint are different address families")]
@@ -217,6 +229,16 @@ mod tests {
             }),
             Some((BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), vec![]))
         );
+        // Something of ours that cannot be written is our fault: the peer
+        // is told the session is being reset, not left to time it out.
+        assert_eq!(
+            notification(SessionError::Encode(EncodeError::TooLong {
+                what: "path attribute",
+                len: 70000,
+                max: 65535
+            })),
+            Some((BgpError::Cease(CeaseSubcode::AdminReset), vec![]))
+        );
         // A broken connection cannot carry a NOTIFICATION.
         assert_eq!(
             notification(SessionError::Io(io::ErrorKind::BrokenPipe.into())),

crates/server/src/fsm.rs

@@ -18,8 +18,6 @@ use bgp_packet::errors::{BgpError, DecodeError, OpenMsgSubcode};
 use bgp_packet::message::{AS_TRANS, BGP4_VERSION, Message, UpdateMessage};
 use bgp_packet::open::{AddPathEntry, AddPathSendRecv, Capability, OpenMessage, OpenOption};
 
-use bgp_rib::table::TooManyPeers;
-
 use crate::error::SessionError;
 use crate::family::FamilySet;
 use crate::metrics::message_type_name;
@@ -73,6 +71,8 @@ pub struct FsmConfig {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Negotiated {
     pub peer_asn: u32,
+    /// Whether the peer is in our own AS.
+    pub ibgp: bool,
     pub peer_identifier: Ipv4Addr,
     /// Zero means the hold and keepalive timers do not run (RFC 4271 §4.2).
     pub hold_time_secs: u16,
@@ -90,11 +90,30 @@ pub struct Negotiated {
     pub add_path_send: FamilySet,
 }
 
+/// A message from the peer, with what the decoder found out besides.
+#[derive(Debug)]
+pub struct Received {
+    pub message: Message,
+    /// For an UPDATE: its attributes as they were on the wire, if the
+    /// decoder kept them (`DecodedUpdate::attributes_wire`). They go with
+    /// the UPDATE to whoever stores its routes.
+    pub attributes_wire: Option<Vec<u8>>,
+}
+
+impl From<Message> for Received {
+    fn from(message: Message) -> Self {
+        Self {
+            message,
+            attributes_wire: None,
+        }
+    }
+}
+
 /// Something that happened to the connection.
 #[derive(Debug)]
 pub enum Event {
     /// The peer sent a message.
-    Message(Message),
+    Message(Received),
     /// The peer sent something that is not a valid message.
     DecodeFailed(DecodeError),
     HoldTimerExpired,
@@ -104,9 +123,10 @@ pub enum Event {
     CollisionLost,
     /// The rest of the server wants this message sent to the peer.
     Outbound(Message),
-    /// The routing table has no room for this peer's routes, so the session
-    /// cannot go on.
-    TableFull(TooManyPeers),
+    /// Something on our side keeps the session from going on: the routing
+    /// table has no room for the peer's routes, say. The peer is told as
+    /// the error has it.
+    LocalFailure(SessionError),
 }
 
 /// Something the driver must do, in the order given.
@@ -128,8 +148,12 @@ pub enum Action {
     /// The peer sent this many routes of address families that were not
     /// negotiated. They have been left out of the update that follows.
     RoutesIgnored(usize),
-    /// Routes arrived; hand them to the RIB.
-    DeliverUpdate(UpdateMessage),
+    /// Routes arrived; put them in the routing table.
+    DeliverUpdate {
+        update: UpdateMessage,
+        /// See [`Received::attributes_wire`].
+        attributes_wire: Option<Vec<u8>>,
+    },
     /// The peer asked for everything we advertise to be sent again.
     RouteRefreshRequested,
     /// The session is over, for this reason. Always the last action.
@@ -179,7 +203,8 @@ impl Fsm {
     }
 
     /// What the OPENs settled; `None` until the peer's OPEN is accepted.
-    pub fn negotiated(&self) -> Option<&Negotiated> {
+    #[cfg(test)]
+    fn negotiated(&self) -> Option<&Negotiated> {
         match &self.phase {
             Phase::OpenConfirm(negotiated) | Phase::Established(negotiated) => Some(negotiated),
             Phase::OpenSent | Phase::Closed => None,
@@ -192,10 +217,10 @@ impl Fsm {
             return vec![];
         }
         match event {
-            Event::Message(msg) => self.on_message(msg),
+            Event::Message(received) => self.on_message(received),
             Event::DecodeFailed(e) => self.fail(SessionError::Decode(e)),
             Event::CollisionLost => self.fail(SessionError::Collision),
-            Event::TableFull(e) => self.fail(SessionError::TableFull(e)),
+            Event::LocalFailure(e) => self.fail(e),
             Event::HoldTimerExpired => self.fail(match self.phase {
                 Phase::Established(_) => SessionError::HoldTimerExpired,
                 _ => SessionError::Timeout(self.state()),
@@ -208,8 +233,12 @@ impl Fsm {
         }
     }
 
-    fn on_message(&mut self, msg: Message) -> Vec<Action> {
-        match (&self.phase, msg) {
+    fn on_message(&mut self, received: Received) -> Vec<Action> {
+        let Received {
+            message,
+            attributes_wire,
+        } = received;
+        match (&self.phase, message) {
             // A NOTIFICATION ends the session in any state, and is not
             // answered (RFC 4271 §8.2.2).
             (_, Message::Notification(notification)) => {
@@ -240,7 +269,10 @@ impl Fsm {
                 if ignored > 0 {
                     actions.push(Action::RoutesIgnored(ignored));
                 }
-                actions.push(Action::DeliverUpdate(update));
+                actions.push(Action::DeliverUpdate {
+                    update,
+                    attributes_wire,
+                });
                 actions
             }
             (Phase::Established(negotiated), Message::RouteRefresh(_)) => {
@@ -304,6 +336,7 @@ impl Fsm {
 
         let negotiated = Negotiated {
             peer_asn,
+            ibgp: peer_asn == self.config.local_asn,
             peer_identifier: open.identifier,
             hold_time_secs: self.config.hold_time_secs.min(open.hold_time),
             four_octet_asn: capability_asn.is_some(),
@@ -428,6 +461,8 @@ mod tests {
     use bgp_primitives::constants::AfiSafi;
     use bgp_primitives::ip_prefix::IpPrefix;
 
+    use bgp_rib::table::TooManyPeers;
+
     use super::*;
     use crate::family::SUPPORTED;
 
@@ -470,15 +505,19 @@ mod tests {
     fn fsm_in(state: State, peer_hold_time: u16) -> Fsm {
         let (mut fsm, _) = Fsm::new(config());
         if state != State::OpenSent {
-            fsm.handle(Event::Message(Message::Open(peer_open(peer_hold_time))));
+            fsm.handle(received(Message::Open(peer_open(peer_hold_time))));
         }
         if state == State::Established {
-            fsm.handle(Event::Message(Message::KeepAlive));
+            fsm.handle(received(Message::KeepAlive));
         }
         assert_eq!(fsm.state(), state);
         fsm
     }
 
+    fn received(message: Message) -> Event {
+        Event::Message(message.into())
+    }
+
     fn empty_update() -> UpdateMessage {
         UpdateMessage {
             withdrawn_routes: vec![],
@@ -525,10 +564,11 @@ mod tests {
         let (mut fsm, _) = Fsm::new(config());
 
         // The peer proposes 30 seconds, we 90: the smaller one holds.
-        let actions = fsm.handle(Event::Message(Message::Open(peer_open(30))));
+        let actions = fsm.handle(received(Message::Open(peer_open(30))));
         assert_eq!(fsm.state(), State::OpenConfirm);
         let expected = Negotiated {
             peer_asn: 65001,
+            ibgp: false,
             peer_identifier: PEER_ID,
             hold_time_secs: 30,
             four_octet_asn: false,
@@ -546,7 +586,7 @@ mod tests {
             ] if *n == expected && *hold == Duration::from_secs(30)
         ));
 
-        let actions = fsm.handle(Event::Message(Message::KeepAlive));
+        let actions = fsm.handle(received(Message::KeepAlive));
         assert_eq!(fsm.state(), State::Established);
         assert!(matches!(
             &actions[..],
@@ -563,20 +603,20 @@ mod tests {
     #[test]
     fn a_hold_time_of_zero_runs_no_timers_once_established() {
         let (mut fsm, _) = Fsm::new(config());
-        let actions = fsm.handle(Event::Message(Message::Open(peer_open(0))));
+        let actions = fsm.handle(received(Message::Open(peer_open(0))));
         // The KEEPALIVE that confirms the session is still waited for.
         assert!(matches!(
             actions.last(),
             Some(Action::ResetHoldTimer(INITIAL_HOLD_TIME))
         ));
 
-        let actions = fsm.handle(Event::Message(Message::KeepAlive));
+        let actions = fsm.handle(received(Message::KeepAlive));
         assert!(matches!(
             &actions[..],
             [Action::Established(_), Action::StopHoldTimer]
         ));
         // And hearing from the peer restarts nothing.
-        assert!(fsm.handle(Event::Message(Message::KeepAlive)).is_empty());
+        assert!(fsm.handle(received(Message::KeepAlive)).is_empty());
     }
 
     #[test]
@@ -592,7 +632,7 @@ mod tests {
             }])],
             ..peer_open(90)
         };
-        fsm.handle(Event::Message(Message::Open(open)));
+        fsm.handle(received(Message::Open(open)));
         let negotiated = fsm.negotiated().unwrap();
         assert_eq!(negotiated.peer_asn, 210036);
         assert!(negotiated.four_octet_asn);
@@ -630,7 +670,7 @@ mod tests {
                 )],
                 ..peer_open(90)
             };
-            let actions = fsm.handle(Event::Message(Message::Open(open)));
+            let actions = fsm.handle(received(Message::Open(open)));
             (fsm, actions)
         };
 
@@ -691,7 +731,7 @@ mod tests {
                 options: vec![OpenOption::Capabilities(capabilities)],
                 ..peer_open(90)
             };
-            fsm.handle(Event::Message(Message::Open(open)));
+            fsm.handle(received(Message::Open(open)));
             let n = fsm.negotiated().unwrap().clone();
             (first, n.add_path_receive, n.add_path_send)
         };
@@ -754,14 +794,14 @@ mod tests {
             announced_routes: vec![v4.into()],
         };
 
-        let actions = fsm.handle(Event::Message(Message::Update(update)));
+        let actions = fsm.handle(received(Message::Update(update)));
         assert_eq!(fsm.state(), State::Established, "the session survives");
         assert!(matches!(
             &actions[..],
             [
                 Action::ResetHoldTimer(_),
                 Action::RoutesIgnored(1),
-                Action::DeliverUpdate(delivered)
+                Action::DeliverUpdate { update: delivered, .. }
             ] if delivered.announced_routes == [v4] && delivered.path_attributes.is_empty()
         ));
     }
@@ -770,7 +810,7 @@ mod tests {
     fn an_unacceptable_open_is_refused_with_the_matching_subcode() {
         let refused = |open: OpenMessage, config: FsmConfig| {
             let (mut fsm, _) = Fsm::new(config);
-            let actions = fsm.handle(Event::Message(Message::Open(open)));
+            let actions = fsm.handle(received(Message::Open(open)));
             ending(&fsm, actions).0
         };
         let open_error = |subcode| Some(BgpError::OpenMsg(subcode));
@@ -833,7 +873,7 @@ mod tests {
         ];
         for (state, msg, subcode) in cases {
             let mut fsm = fsm_in(state, 90);
-            let actions = fsm.handle(Event::Message(msg));
+            let actions = fsm.handle(received(msg));
             let (sent, result) = ending(&fsm, actions);
             assert_eq!(sent, Some(BgpError::FsmError(subcode)), "{state}");
             assert!(
@@ -899,7 +939,7 @@ mod tests {
         for state in [State::OpenSent, State::OpenConfirm, State::Established] {
             let mut fsm = fsm_in(state, 90);
             let cease = BgpError::Cease(CeaseSubcode::AdminShutdown);
-            let actions = fsm.handle(Event::Message(Message::Notification(NotificationMessage {
+            let actions = fsm.handle(received(Message::Notification(NotificationMessage {
                 error: cease.clone(),
                 data: vec![],
             })));
@@ -914,16 +954,16 @@ mod tests {
         let mut fsm = fsm_in(State::Established, 30);
         let hold = Duration::from_secs(30);
 
-        let actions = fsm.handle(Event::Message(Message::KeepAlive));
+        let actions = fsm.handle(received(Message::KeepAlive));
         assert!(matches!(&actions[..], [Action::ResetHoldTimer(d)] if *d == hold));
 
-        let actions = fsm.handle(Event::Message(Message::Update(empty_update())));
+        let actions = fsm.handle(received(Message::Update(empty_update())));
         assert!(matches!(
             &actions[..],
-            [Action::ResetHoldTimer(d), Action::DeliverUpdate(_)] if *d == hold
+            [Action::ResetHoldTimer(d), Action::DeliverUpdate { .. }] if *d == hold
         ));
 
-        let actions = fsm.handle(Event::Message(Message::RouteRefresh(
+        let actions = fsm.handle(received(Message::RouteRefresh(
             RouteRefreshMessage::request(AfiSafi::IPV4_UNICAST),
         )));
         assert!(matches!(
@@ -970,7 +1010,7 @@ mod tests {
     #[test]
     fn a_full_table_ends_the_session_and_says_why() {
         let mut fsm = fsm_in(State::Established, 90);
-        let actions = fsm.handle(Event::TableFull(TooManyPeers));
+        let actions = fsm.handle(Event::LocalFailure(SessionError::TableFull(TooManyPeers)));
         let (sent, result) = ending(&fsm, actions);
         assert_eq!(sent, Some(BgpError::Cease(CeaseSubcode::OutOfResources)));
         assert!(matches!(result, Err(SessionError::TableFull(_))));
@@ -982,7 +1022,7 @@ mod tests {
         fsm.handle(Event::HoldTimerExpired);
         assert_eq!(fsm.state(), State::Closed);
         for event in [
-            Event::Message(Message::KeepAlive),
+            received(Message::KeepAlive),
             Event::HoldTimerExpired,
             Event::KeepaliveTimerExpired,
             Event::CollisionLost,

crates/server/src/server.rs

@@ -1,3 +1,6 @@
+//! The server: listeners, the tasks that run for as long as it does, and
+//! what happens when one of them stops.
+
 use std::io;
 use std::net::SocketAddr;
 use std::pin::Pin;
@@ -5,7 +8,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use futures::stream::{FuturesUnordered, StreamExt};
-use tokio::net::{TcpListener, TcpStream};
+use tokio::net::{TcpListener, TcpSocket, TcpStream};
 use tokio::task::{JoinError, JoinHandle};
 use tracing::{error, info, warn};
 
@@ -15,19 +18,30 @@ use bgp_rib::{LocalRib, RibStopped};
 use crate::advertise::spawn_advertiser;
 use crate::collision::{CollisionTable, Direction};
 use crate::config::Config;
-use crate::connection::{SessionConfig, handle_connection};
+use crate::connection::{PeerSettings, ServerContext, run_session};
 use crate::dialer::spawn_dialers;
 use crate::error::ServerError;
-use crate::family::FamilySet;
 use crate::panic;
 use crate::session::{ConfiguredPeer, SessionRegistry};
 
-pub struct BgpServerBuilder {
-    config: Config,
-}
+/// How long sessions get to send their Cease when the server stops.
+const SHUTDOWN_GRACE: Duration = Duration::from_secs(2);
 
-/// A `BgpServerBuilder` that has bound all its listeners and is ready to accept.
-/// Call [`BgpServer::run`] to start the accept loop.
+/// How long to wait before accepting again after an accept failed. What
+/// makes it fail (no file descriptors left, usually) does not pass at once,
+/// and trying again in a loop would only fill the log.
+const ACCEPT_RETRY: Duration = Duration::from_millis(200);
+
+/// How often attribute sets that no route uses any more are dropped. They
+/// are not dropped as their last route goes (see `AttrStore::sweep`), and a
+/// table whose routes keep changing attributes collects them.
+const SWEEP_INTERVAL: Duration = Duration::from_secs(60);
+
+/// How many connections may wait to be accepted on a listener.
+const LISTEN_BACKLOG: u32 = 1024;
+
+/// A BGP speaker that has bound its listeners and is ready to accept.
+/// Call [`BgpServer::run`] to start it.
 pub struct BgpServer {
     config: Config,
     listeners: Vec<(SocketAddr, TcpListener)>,
@@ -41,20 +55,14 @@ pub struct BgpServer {
     pub registry: SessionRegistry,
 }
 
-impl BgpServerBuilder {
-    pub fn new(config: Config) -> Self {
-        Self { config }
-    }
-
-    /// Binds all configured listen endpoints and returns a [`BgpServer`].
-    /// Separating bind from run lets callers (e.g. tests) inspect the actual
-    /// bound addresses before the accept loop starts.
-    pub async fn bind(self) -> Result<BgpServer, ServerError> {
-        let mut listeners = Vec::with_capacity(self.config.listen_endpoints.len());
-        for &addr in &self.config.listen_endpoints {
-            let listener = TcpListener::bind(addr)
-                .await
-                .map_err(|source| ServerError::Bind { addr, source })?;
+impl BgpServer {
+    /// Binds all configured listen endpoints. Separate from
+    /// [`BgpServer::run`] so that callers (tests, say) can learn the
+    /// addresses actually bound before any connection is accepted.
+    pub async fn bind(config: Config) -> Result<Self, ServerError> {
+        let mut listeners = Vec::with_capacity(config.listen_endpoints.len());
+        for &addr in &config.listen_endpoints {
+            let listener = listen(addr).map_err(|source| ServerError::Bind { addr, source })?;
             let bound_addr = listener.local_addr()?;
             info!(%bound_addr, "Listening for BGP connections");
             listeners.push((bound_addr, listener));
@@ -62,15 +70,10 @@ impl BgpServerBuilder {
         // Routes from the configuration go in before any session exists;
         // each session is sent the whole table when it comes up.
         let local_rib = LocalRib::spawn();
-        for route in &self.config.originate {
+        for route in &config.originate {
             local_rib.announce(route.clone()).await?;
         }
-        Ok(BgpServer {
-            listeners,
-            rib: Arc::new(RibTable::new(table_shards())),
-            local_rib,
-            registry: SessionRegistry::with_configured(
-                self.config
+        let configured = config
             .peers
             .iter()
             .map(|p| ConfiguredPeer {
@@ -78,19 +81,16 @@ impl BgpServerBuilder {
                 endpoint: p.endpoint,
                 remote_asn: p.remote_asn,
             })
-                    .collect(),
-            ),
-            config: self.config,
+            .collect();
+        Ok(BgpServer {
+            listeners,
+            rib: Arc::new(RibTable::new(table_shards())),
+            local_rib,
+            registry: SessionRegistry::with_configured(configured),
+            config,
         })
     }
 
-    /// Convenience method: bind then run immediately.
-    pub async fn run(self) -> Result<(), ServerError> {
-        self.bind().await?.run().await
-    }
-}
-
-impl BgpServer {
     pub fn config(&self) -> &Config {
         &self.config
     }
@@ -100,6 +100,8 @@ impl BgpServer {
         self.listeners.iter().map(|(addr, _)| *addr).collect()
     }
 
+    /// Runs the server. Returns only if something it cannot do without has
+    /// failed, after closing its sessions.
     pub async fn run(self) -> Result<(), ServerError> {
         crate::metrics::init();
         info!(
@@ -109,30 +111,18 @@ impl BgpServer {
             "BGP server started"
         );
 
-        let session_config = SessionConfig {
+        let context = Arc::new(ServerContext {
             router_id: self.config.router_id,
             local_asn: self.config.local_asn,
-            hold_time_secs: self.config.hold_time_secs,
             rib: Arc::clone(&self.rib),
             registry: self.registry.clone(),
             collisions: CollisionTable::new(self.config.collision_detect_established_state),
-            direction: Direction::Inbound,
-            peer_name: None,
-            expected_asn: None,
-            families: FamilySet::supported(),
-            add_path: None,
-        };
+        });
 
         // The tasks the server cannot do without. None of them ends while
-        // the server runs, unless it panics.
-        let mut tasks: FuturesUnordered<_> = spawn_dialers(
-            &self.config.peers,
-            &session_config,
-            self.config.connect_retry,
-        )
-        .into_iter()
-        .map(|dialer| named("dialer", dialer))
-        .collect();
+        // the server runs. The advertiser comes first: it has to be
+        // listening before a dialer brings up the first session.
+        let mut tasks: FuturesUnordered<NamedTask> = FuturesUnordered::new();
         tasks.push(named(
             "advertiser",
             spawn_advertiser(
@@ -141,6 +131,10 @@ impl BgpServer {
                 self.config.local_asn,
             ),
         ));
+        tasks.push(named("sweeper", spawn_sweeper(Arc::clone(&self.rib))));
+        for dialer in spawn_dialers(&self.config.peers, &context, self.config.connect_retry) {
+            tasks.push(named("dialer", dialer));
+        }
 
         let mut accepting: FuturesUnordered<_> = self
             .listeners
@@ -152,31 +146,24 @@ impl BgpServer {
             let (listen_addr, result, listener) = tokio::select! {
                 Some(accepted) = accepting.next() => accepted,
                 failure = failed(&self.local_rib, &mut tasks) => {
-                    return Err(give_up(&self.registry, failure).await);
+                    error!(err = %failure, "Shutting down: the server cannot continue");
+                    close_sessions(&self.registry).await;
+                    return Err(failure);
                 }
             };
             match result {
                 Ok((stream, peer_addr)) => {
-                    // Inbound connections from a configured peer inherit its
-                    // name, expected ASN, hold time, address families and
-                    // ADD-PATH setting.
-                    let mut config = session_config.clone();
-                    if let Some(peer) = self
-                        .config
-                        .peers
-                        .iter()
-                        .find(|p| p.endpoint.ip() == peer_addr.ip())
-                    {
-                        config.peer_name = Some(peer.name.clone());
-                        config.expected_asn = Some(peer.remote_asn);
-                        config.hold_time_secs = peer.hold_time_secs;
-                        config.families = peer.families.clone();
-                        config.add_path = peer.add_path;
-                    }
-                    tokio::spawn(handle_connection(stream, peer_addr, Arc::new(config)));
+                    // A connection from a configured peer is run with what
+                    // is configured for that peer.
+                    let peer = self.config.peer_at(peer_addr.ip()).map_or_else(
+                        || PeerSettings::unconfigured(self.config.hold_time_secs),
+                        PeerSettings::from,
+                    );
+                    tokio::spawn(serve(stream, peer_addr, Arc::clone(&context), peer));
                 }
                 Err(e) => {
                     error!(%listen_addr, err = %e, "Accept error");
+                    tokio::time::sleep(ACCEPT_RETRY).await;
                 }
             }
             accepting.push(accept_one(listen_addr, listener));
@@ -184,17 +171,60 @@ impl BgpServer {
     }
 }
 
-/// Ends the server because a part of it that it cannot do without has
-/// failed. Peers are told we are going (Cease / Administrative Shutdown),
-/// so that they stop using our routes now and not when their hold timers
-/// expire.
-async fn give_up(registry: &SessionRegistry, failure: ServerError) -> ServerError {
-    error!(err = %failure, "Shutting down: the server cannot continue");
+/// 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;
     if left > 0 {
         warn!(sessions = left, "Some sessions did not close in time");
     }
-    failure
+}
+
+/// Opens a listening socket on `addr`.
+fn listen(addr: SocketAddr) -> io::Result<TcpListener> {
+    let socket = match addr {
+        SocketAddr::V4(_) => TcpSocket::new_v4()?,
+        SocketAddr::V6(_) => {
+            let socket = TcpSocket::new_v6()?;
+            // IPv6 only. Where a system's default is to take IPv4
+            // connections on such a socket too, listening on both 0.0.0.0
+            // and [::] would fail on the second, and IPv4 peers would come
+            // in under IPv4-mapped addresses that match no configured peer.
+            socket2::SockRef::from(&socket).set_only_v6(true)?;
+            socket
+        }
+    };
+    socket.set_reuseaddr(true)?;
+    socket.bind(addr)?;
+    socket.listen(LISTEN_BACKLOG)
+}
+
+/// Runs a session on a connection a peer opened, and logs how it ended.
+async fn serve(
+    stream: TcpStream,
+    peer_addr: SocketAddr,
+    context: Arc<ServerContext>,
+    peer: PeerSettings,
+) {
+    let end = run_session(stream, peer_addr, Direction::Inbound, context, peer).await;
+    if let Err(e) = end.result {
+        warn!(%peer_addr, err = %e, "BGP session terminated with error");
+    }
+}
+
+/// Drops, every so often, the attribute sets no route uses any more.
+fn spawn_sweeper(rib: Arc<RibTable>) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        let mut tick = tokio::time::interval(SWEEP_INTERVAL);
+        loop {
+            tick.tick().await;
+            let rib = 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 || rib.attrs().sweep()).await;
+        }
+    })
 }
 
 /// How many shards the routing table has. Sessions write to it from every
@@ -204,9 +234,6 @@ fn table_shards() -> usize {
     16 * std::thread::available_parallelism().map_or(1, |cores| cores.get())
 }
 
-/// How long sessions get to send their Cease when the server gives up.
-const SHUTDOWN_GRACE: Duration = Duration::from_secs(2);
-
 type NamedTask = Pin<Box<dyn Future<Output = (&'static str, Result<(), JoinError>)> + Send>>;
 
 fn named(name: &'static str, task: JoinHandle<()>) -> NamedTask {

crates/server/src/session.rs

@@ -15,7 +15,6 @@ use bgp_packet::errors::{BgpError, CeaseSubcode};
 use bgp_packet::message::NotificationMessage;
 
 use bgp_packet::message::Message;
-use bgp_rib::PeerId;
 use bgp_rib::table::PeerIndex;
 use tokio::sync::{broadcast, mpsc};
 
@@ -27,6 +26,19 @@ use crate::sync::IgnorePoisonRw;
 
 const EVENT_CHANNEL_CAPACITY: usize = 256;
 
+/// Identifies a session by the peer's TCP address and its AS number.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct PeerId {
+    pub addr: SocketAddr,
+    pub asn: u32,
+}
+
+impl std::fmt::Display for PeerId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}(AS{})", self.addr, self.asn)
+    }
+}
+
 /// Where the dialer is in bringing up a configured peer.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum DialState {
@@ -36,6 +48,7 @@ pub enum DialState {
     /// The last attempt failed or the session ended; waiting to retry.
     Backoff {
         until: SystemTime,
+        /// How the last attempt or session ended.
         last_error: String,
     },
 }
@@ -60,6 +73,8 @@ pub struct SessionHandle {
     /// originate to this peer.
     pub local_addr: SocketAddr,
     pub established_at: SystemTime,
+    /// Whether the peer is in our own AS.
+    pub ibgp: bool,
     /// Whether the peer negotiated RFC 6793 four-octet ASNs. Determines how
     /// AS_PATH is encoded on the wire for this session.
     pub four_octet_asn: bool,
@@ -82,7 +97,7 @@ pub struct SessionHandle {
 pub struct SessionClosed(pub PeerId);
 
 impl SessionHandle {
-    pub fn new(
+    pub(crate) fn new(
         peer: PeerId,
         name: Option<String>,
         direction: Direction,
@@ -97,6 +112,7 @@ impl SessionHandle {
             direction,
             local_addr,
             established_at: SystemTime::now(),
+            ibgp: negotiated.ibgp,
             four_octet_asn: negotiated.four_octet_asn,
             families: negotiated.families.clone(),
             add_path_receive: negotiated.add_path_receive.clone(),
@@ -176,12 +192,7 @@ impl SessionRegistry {
             .collect()
     }
 
-    /// The configured peer whose endpoint IP is `ip`, if any.
-    pub fn configured_by_ip(&self, ip: IpAddr) -> Option<&ConfiguredPeer> {
-        self.configured.iter().find(|p| p.endpoint.ip() == ip)
-    }
-
-    pub fn set_dial_state(&self, name: &str, state: DialState) {
+    pub(crate) fn set_dial_state(&self, name: &str, state: DialState) {
         self.dial_states.writing().insert(name.to_string(), state);
     }
 
@@ -197,7 +208,7 @@ impl SessionRegistry {
 
     /// Adds a session and emits [`SessionEvent::Up`]. The returned guard
     /// removes the session again when dropped.
-    pub fn register(&self, handle: SessionHandle) -> SessionGuard {
+    pub(crate) fn register(&self, handle: SessionHandle) -> SessionGuard {
         let peer = handle.peer;
         self.inner.writing().insert(peer, handle.clone());
         SESSIONS_ESTABLISHED.inc();
@@ -208,18 +219,15 @@ impl SessionRegistry {
         }
     }
 
-    pub fn get(&self, peer: &PeerId) -> Option<SessionHandle> {
-        self.inner.reading().get(peer).cloned()
-    }
-
     /// All currently established sessions, in no particular order.
     pub fn list(&self) -> Vec<SessionHandle> {
         self.inner.reading().values().cloned().collect()
     }
 
     /// Called by a session when its peer asks for a route refresh.
-    pub fn request_refresh(&self, peer: &PeerId) {
-        if let Some(handle) = self.get(peer) {
+    pub(crate) fn request_refresh(&self, peer: &PeerId) {
+        let handle = self.inner.reading().get(peer).cloned();
+        if let Some(handle) = handle {
             let _ = self.events.send(SessionEvent::RefreshRequested(handle));
         }
     }
@@ -237,12 +245,15 @@ impl SessionRegistry {
     pub async fn shutdown(&self, grace: Duration) -> usize {
         self.shutting_down.store(true, Ordering::Relaxed);
         for session in self.list() {
+            // Not waited for: a session whose queue is full is one whose
+            // peer is not reading, and it would hold up the goodbye to all
+            // the others. It ends when the process does.
             let _ = session
-                .send(Message::Notification(NotificationMessage {
+                .outbound
+                .try_send(Message::Notification(NotificationMessage {
                     error: BgpError::Cease(CeaseSubcode::AdminShutdown),
                     data: vec![],
-                }))
-                .await;
+                }));
         }
         let deadline = tokio::time::Instant::now() + grace;
         while !self.list().is_empty() && tokio::time::Instant::now() < deadline {

crates/server/src/tracing.rs

@@ -8,12 +8,10 @@ const DEFAULT_TARGETS: [&str; 3] = ["bgp", "bgp_server", "bgp_web"];
 
 #[derive(Debug, Args)]
 pub struct TracingArgs {
-    /// Set the minimum log level.
-    #[arg(short, long, global = true, default_value_t = Level::TRACE, help_heading = "Tracing")]
+    /// Set the minimum log level. At `debug` and below there is a line for
+    /// every UPDATE and every route announced.
+    #[arg(short, long, global = true, default_value_t = Level::INFO, help_heading = "Tracing")]
     level: Level,
-    /// Disable logging to stdout.
-    #[arg(long, global = true, default_value_t = false, help_heading = "Tracing")]
-    quiet: bool,
 }
 
 impl TracingArgs {

crates/web/src/api.rs

@@ -12,8 +12,8 @@ use axum::response::{IntoResponse, Response};
 use bgp_packet::community;
 use bgp_packet::message::{NlriNextHop, OriginPathAttribute, PathAttribute};
 use bgp_rib::table::{PeerIndex, Route};
-use bgp_rib::{LocalRoute, PeerId, RibStopped};
-use bgp_server::session::{ConfiguredPeer, DialState, SessionHandle, SessionRegistry};
+use bgp_rib::{LocalRoute, RibStopped};
+use bgp_server::session::{ConfiguredPeer, DialState, PeerId, SessionHandle, SessionRegistry};
 use serde::{Deserialize, Serialize};
 
 use crate::AppState;

crates/web/src/events.rs

@@ -8,7 +8,7 @@ use axum::extract::State;
 use axum::response::sse::{Event, KeepAlive, Sse};
 use bgp_primitives::ip_prefix::IpPrefix;
 use bgp_rib::LocalRibEvent;
-use bgp_rib::table::{self as journal, Route};
+use bgp_rib::table::{self, Route};
 use bgp_server::session::SessionEvent;
 use futures::stream::{Stream, StreamExt};
 use serde::Serialize;
@@ -116,12 +116,12 @@ fn best_routes(state: &AppState) -> impl Stream<Item = WireEvent> + use<> {
                 // the views are made afterwards.
                 let mut changes = Vec::new();
                 let progress = reader.poll(BEST_ROUTE_BUFFER, |event| match event {
-                    journal::Event::Reset => {
+                    table::Event::Reset => {
                         syncing = true;
                         changes.push(Change::Lagged);
                     }
-                    journal::Event::Synced => syncing = false,
-                    journal::Event::Best {
+                    table::Event::Synced => syncing = false,
+                    table::Event::Best {
                         prefix,
                         peer,
                         path_id,
@@ -135,7 +135,7 @@ fn best_routes(state: &AppState) -> impl Stream<Item = WireEvent> + use<> {
                             best: true,
                         }));
                     }
-                    journal::Event::Withdrawn { prefix } if !syncing => {
+                    table::Event::Withdrawn { prefix } if !syncing => {
                         changes.push(Change::Withdrawn(prefix));
                     }
                     _ => {}

crates/web/src/lib.rs

@@ -2,7 +2,7 @@
 //! status page, a server-sent event stream, and Prometheus metrics.
 //!
 //! Everything here is a thin layer over handles owned by the BGP server
-//! ([`SessionRegistry`], [`LocalRib`], [`AdjRibIn`]); no routing state lives
+//! ([`SessionRegistry`], [`LocalRib`], [`RibTable`]); no routing state lives
 //! in this crate.
 
 mod api;