Make the session state machine explicit and free of I/O

A session was one function that walked OpenSent, OpenConfirm and
Established with the socket, the timers, collision detection, the registry
and the RIB all in the same loop. There was no state or event type, and no
way to test a transition without a TCP connection.

It is now two pieces:

- fsm.rs is the state machine of RFC 4271 section 8 for a connection. It
  takes an Event and returns the Actions to carry out. It owns no socket,
  timer or task. What the OPENs negotiated lives in the states that have
  it, so there is nothing to unwrap. Thirteen tests cover the handshake,
  hold-time negotiation, refused OPENs, messages out of turn, timer expiry,
  collisions and shutdown by feeding it events.
- connection.rs is the driver: it turns the socket, the timers and the
  outbound queue into events and carries the actions out.

Behaviour changes:

- A message that arrived in pieces was lost if a timer fired between the
  pieces: read_exact was used inside select!, which discards what it has
  read when another branch wins. Messages are now framed from a buffer
  that survives cancellation.
- A NOTIFICATION from the peer during the handshake is no longer answered
  with an FSM error; RFC 4271 says to close without replying.
- A session the peer ended is reported as 'peer sent NOTIFICATION: ...'
  with its error, where the dialer used to record 'session closed by peer'.

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

crates/integration-tests/tests/basic_connection.rs

@@ -180,6 +180,49 @@ async fn test_malformed_messages_are_answered_with_the_matching_notification() -
     Ok(())
 }
 
+/// A message may arrive in pieces, with the server's own timers firing in
+/// between. The server once threw away the piece it had already read when
+/// that happened, so the message was silently lost (and the stream was only
+/// back in step if the remainder happened to be thrown away as well).
+#[tokio::test]
+async fn test_message_split_across_a_keepalive_tick_is_still_one_message() -> Result<()> {
+    // Hold time 3 s: the server sends a KEEPALIVE every second.
+    let server = TestServer::start_with_hold_time(65000, 3).await?;
+    let mut peer = TestPeer::connect(server.addr).await?;
+    peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 3).await?;
+
+    let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?;
+    let mut update = bytes::BytesMut::new();
+    Message::Update(UpdateMessage {
+        withdrawn_routes: vec![],
+        path_attributes: vec![],
+        announced_routes: vec![prefix.clone()],
+    })
+    .to_wire(&peer.ctx, &mut update)?;
+
+    // The first half, a pause long enough for the server's keepalive timer
+    // to fire, then the rest.
+    peer.send_raw(&update[..10]).await?;
+    tokio::time::sleep(Duration::from_millis(1300)).await;
+    peer.send_raw(&update[10..]).await?;
+
+    tokio::time::timeout(TIMEOUT, async {
+        while !server
+            .adj_rib_in
+            .snapshot()
+            .await
+            .iter()
+            .any(|e| e.prefix == prefix)
+        {
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    })
+    .await
+    .map_err(|_| eyre::eyre!("the split UPDATE never reached the RIB"))?;
+    assert_eq!(server.registry.list().len(), 1, "session was dropped");
+    Ok(())
+}
+
 // ---------------------------------------------------------------------------
 // OpenConfirm → Established
 // ---------------------------------------------------------------------------

crates/server/src/connection.rs

@@ -1,11 +1,61 @@
-use std::net::SocketAddr;
+//! Drives one TCP connection through the BGP state machine.
+//!
+//! [`crate::fsm::Fsm`] decides; this module does. It turns what arrives on
+//! the socket, the timers and the rest of the server into
+//! [`Event`]s, and carries out the [`Action`]s that come back: writing to
+//! the socket, arming timers, registering the session, feeding the RIB.
+
+use std::collections::VecDeque;
+use std::future::pending;
+use std::io;
+use std::net::{Ipv4Addr, SocketAddr};
+use std::pin::Pin;
 use std::sync::Arc;
 
+use bgp_packet::errors::{BgpError, DecodeError, MsgHeaderSubcode};
+use bgp_packet::message::{BGP_HEADER_LEN, BGP_MAX_MSG_LEN, Message, PathAttribute, UpdateMessage};
+use bgp_primitives::parser::ParserContext;
+use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate};
+use bytes::BytesMut;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
-use tracing::warn;
+use tokio::net::tcp::OwnedWriteHalf;
+use tokio::sync::mpsc;
+use tokio::time::{Instant, Interval, Sleep, interval_at, sleep};
+use tracing::{info, warn};
 
-use crate::fsm::{SessionConfig, run_session};
+use crate::collision::{CollisionTable, ConnectionGuard, Direction};
+use crate::error::SessionError;
+use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated};
+use crate::metrics::{
+    MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, message_type_name,
+};
+use crate::session::{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 {
+    pub router_id: Ipv4Addr,
+    pub local_asn: u32,
+    pub hold_time_secs: u16,
+    pub adj_rib_in: AdjRibIn,
+    pub registry: SessionRegistry,
+    /// Shared table 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>,
+    /// 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>,
+}
+
+/// Runs a session on an accepted connection and logs how it ended.
 pub async fn handle_connection(
     stream: TcpStream,
     peer_addr: SocketAddr,
@@ -15,3 +65,428 @@ pub async fn handle_connection(
         warn!(%peer_addr, err = %e, "BGP session terminated with error");
     }
 }
+
+/// Runs the BGP session on `stream` until it ends.
+pub async fn run_session(
+    stream: TcpStream,
+    peer_addr: SocketAddr,
+    config: SessionConfig,
+) -> Result<(), SessionError> {
+    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,
+    });
+    let mut session = Session {
+        fsm,
+        frames: FrameReader::new(reader),
+        writer,
+        ctx: ParserContext::default(),
+        hold_timer: None,
+        keepalive_timer: None,
+        collision: None,
+        established: None,
+        peer_addr,
+        local_addr,
+        config,
+    };
+
+    let result = session.run(first_actions).await;
+
+    // Whatever ended the session, routes learned over it are gone
+    // (RFC 4271 §8.2.2). Do this before the registration drops so
+    // subscribers see the withdrawals before the Down event.
+    if let Some(established) = session.established.take() {
+        let removed = session
+            .config
+            .adj_rib_in
+            .remove_peer(established.peer_id)
+            .await;
+        info!(peer = %established.peer_id, removed_routes = removed, "Session ended");
+    }
+    result
+}
+
+/// What exists only once a session is Established.
+struct Established {
+    peer_id: PeerId,
+    /// Messages the rest of the server queued for the peer.
+    outbound: mpsc::Receiver<Message>,
+    /// Keeps the session listed in the registry.
+    _registration: SessionGuard,
+}
+
+struct Session<R> {
+    fsm: Fsm,
+    frames: FrameReader<R>,
+    writer: OwnedWriteHalf,
+    /// How messages are encoded and decoded: the width of AS numbers is
+    /// known once the OPENs are exchanged.
+    ctx: ParserContext,
+    hold_timer: Option<Pin<Box<Sleep>>>,
+    keepalive_timer: Option<Interval>,
+    /// This connection's slot in the collision table, once admitted.
+    collision: Option<ConnectionGuard>,
+    established: Option<Established>,
+    peer_addr: SocketAddr,
+    local_addr: SocketAddr,
+    config: SessionConfig,
+}
+
+impl<R: AsyncRead + Unpin> Session<R> {
+    async fn run(&mut self, first_actions: Vec<Action>) -> Result<(), SessionError> {
+        let mut actions = first_actions;
+        loop {
+            if let Some(result) = self.execute(actions).await {
+                return result;
+            }
+            let event = self.next_event().await?;
+            actions = self.fsm.handle(event);
+        }
+    }
+
+    /// Waits for whatever happens next. Every branch is cancel-safe: the
+    /// branches that lose leave nothing half done.
+    async fn next_event(&mut self) -> Result<Event, SessionError> {
+        tokio::select! {
+            frame = self.frames.next(&self.ctx) => match frame {
+                Ok(msg) => Ok(Event::Message(msg)),
+                Err(SessionError::Decode(e)) => Ok(Event::DecodeFailed(e)),
+                Err(e) => Err(e),
+            },
+            () = expired(&mut self.hold_timer) => Ok(Event::HoldTimerExpired),
+            () = ticked(&mut self.keepalive_timer) => Ok(Event::KeepaliveTimerExpired),
+            msg = queued(&mut self.established) => Ok(Event::Outbound(msg)),
+            () = killed(&mut self.collision) => Ok(Event::CollisionLost),
+        }
+    }
+
+    /// Carries out `actions` in order. Returns the session's result once it
+    /// is over.
+    async fn execute(&mut self, actions: Vec<Action>) -> Option<Result<(), SessionError>> {
+        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));
+                        }
+                        // The session is ending anyway and the Close that
+                        // follows says why; a NOTIFICATION is best effort.
+                        warn!(peer_addr = %self.peer_addr, err = %e, "Failed to send NOTIFICATION");
+                    }
+                }
+                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))),
+                },
+                Action::StopHoldTimer => self.hold_timer = None,
+                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));
+                }
+                Action::Admit(negotiated) => {
+                    self.ctx.four_octet_asn = Some(negotiated.four_octet_asn);
+                    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,
+                                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");
+                            queue = self.fsm.handle(Event::CollisionLost).into();
+                        }
+                    }
+                }
+                Action::Established(negotiated) => self.establish(negotiated),
+                Action::DeliverUpdate(update) => {
+                    if let Some(established) = &self.established {
+                        let (withdrawn, announced) = prefix_counts(&update);
+                        info!(peer = %established.peer_id, withdrawn, announced, "Received UPDATE");
+                        self.config
+                            .adj_rib_in
+                            .apply(established.peer_id, ReceivedUpdate::from_message(&update))
+                            .await;
+                    }
+                }
+                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);
+                    }
+                }
+                Action::Close(result) => return Some(result),
+            }
+        }
+        None
+    }
+
+    /// Makes the session visible to the rest of the server.
+    fn establish(&mut self, negotiated: Negotiated) {
+        let peer_id = PeerId {
+            addr: self.peer_addr,
+            asn: negotiated.peer_asn,
+        };
+        info!(peer_addr = %self.peer_addr, peer_asn = negotiated.peer_asn, "Session established");
+        SESSIONS_ESTABLISHED_TOTAL.inc();
+
+        if let Some(collision) = &mut self.collision {
+            collision.set_established();
+        }
+        let (outbound_tx, outbound) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY);
+        let registration = self.config.registry.register(SessionHandle::new(
+            peer_id,
+            self.config.peer_name.clone(),
+            self.config.direction,
+            self.local_addr,
+            negotiated.four_octet_asn,
+            outbound_tx,
+        ));
+        self.established = Some(Established {
+            peer_id,
+            outbound,
+            _registration: registration,
+        });
+    }
+
+    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);
+            info!(peer = %established.peer_id, withdrawn, announced, "Sending UPDATE");
+        }
+        if let Message::Notification(notification) = msg {
+            info!(peer_addr = %self.peer_addr, error = %notification.error, "Sending NOTIFICATION, closing session");
+        }
+        let mut buf = BytesMut::new();
+        msg.to_wire(&self.ctx, &mut buf)?;
+        self.writer.write_all(&buf).await?;
+        MESSAGES_SENT
+            .with_label_values(&[message_type_name(msg)])
+            .inc();
+        Ok(())
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Event sources. Each waits forever when its source does not exist yet, so
+// that it can sit in the select! from the start.
+// ---------------------------------------------------------------------------
+
+async fn expired(timer: &mut Option<Pin<Box<Sleep>>>) {
+    match timer {
+        Some(timer) => timer.await,
+        None => pending().await,
+    }
+}
+
+async fn ticked(timer: &mut Option<Interval>) {
+    match timer {
+        Some(timer) => {
+            timer.tick().await;
+        }
+        None => pending().await,
+    }
+}
+
+async fn queued(established: &mut Option<Established>) -> Message {
+    match established {
+        // The registry holds the sending half for as long as the session is
+        // registered, so the queue does not close under us; if it somehow
+        // did, there is simply nothing more to send.
+        Some(established) => match established.outbound.recv().await {
+            Some(msg) => msg,
+            None => pending().await,
+        },
+        None => pending().await,
+    }
+}
+
+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;
+        }
+        None => pending().await,
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Framing
+// ---------------------------------------------------------------------------
+
+/// Reads whole BGP messages from a byte stream.
+///
+/// Bytes are collected in a buffer until a full message is there, so
+/// [`FrameReader::next`] can be dropped at any point and called again
+/// without losing data. That matters because it shares a `select!` with the
+/// timers: a message that arrives in two pieces must survive a keepalive
+/// timer firing in between.
+struct FrameReader<R> {
+    reader: R,
+    buf: BytesMut,
+}
+
+impl<R: AsyncRead + Unpin> FrameReader<R> {
+    fn new(reader: R) -> Self {
+        Self {
+            reader,
+            buf: BytesMut::with_capacity(usize::from(BGP_MAX_MSG_LEN)),
+        }
+    }
+
+    /// The next message from the peer. Cancel-safe.
+    async fn next(&mut self, ctx: &ParserContext) -> Result<Message, SessionError> {
+        loop {
+            if let Some(length) = frame_length(&self.buf)?
+                && self.buf.len() >= length
+            {
+                let frame = self.buf.split_to(length);
+                let msg = Message::decode(ctx, &frame)?;
+                MESSAGES_RECEIVED
+                    .with_label_values(&[message_type_name(&msg)])
+                    .inc();
+                return Ok(msg);
+            }
+            if self.reader.read_buf(&mut self.buf).await? == 0 {
+                return Err(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    "peer closed the connection",
+                )
+                .into());
+            }
+        }
+    }
+}
+
+/// The length of the message at the front of `buf`, once its header has
+/// arrived. The length decides how much more to wait for, so it is checked
+/// here, before the rest of the message exists (RFC 4271 §6.1).
+fn frame_length(buf: &[u8]) -> Result<Option<usize>, DecodeError> {
+    if buf.len() < usize::from(BGP_HEADER_LEN) {
+        return Ok(None);
+    }
+    let length = u16::from_be_bytes([buf[16], buf[17]]);
+    if !(BGP_HEADER_LEN..=BGP_MAX_MSG_LEN).contains(&length) {
+        return Err(DecodeError {
+            notification: BgpError::MsgHeader(MsgHeaderSubcode::BadMessageLength),
+            data: length.to_be_bytes().to_vec(),
+            detail: format!(
+                "BGP message length {length} is not between {BGP_HEADER_LEN} and {BGP_MAX_MSG_LEN}"
+            ),
+        });
+    }
+    Ok(Some(usize::from(length)))
+}
+
+/// Prefixes withdrawn and announced by an UPDATE, across the IPv4 fields in
+/// the message body and the multiprotocol attributes (where IPv6 travels).
+fn prefix_counts(update: &UpdateMessage) -> (usize, usize) {
+    let mut withdrawn = update.withdrawn_routes.len();
+    let mut announced = update.announced_routes.len();
+    for attr in &update.path_attributes {
+        match attr {
+            PathAttribute::MpReachNlri(mp) => announced += mp.prefixes.len(),
+            PathAttribute::MpUnreachNlri(mp) => withdrawn += mp.prefixes.len(),
+            _ => {}
+        }
+    }
+    (withdrawn, announced)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use tokio::io::{AsyncWriteExt, duplex};
+
+    use super::*;
+
+    const KEEPALIVE: [u8; 19] = [
+        0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+        0xff, 0x00, 0x13, 0x04,
+    ];
+
+    #[tokio::test]
+    async fn messages_are_split_out_of_the_stream_however_they_arrive() {
+        let (mut peer, ours) = duplex(4096);
+        let mut frames = FrameReader::new(ours);
+        let ctx = ParserContext::default();
+
+        // Two and a half messages in one write, the rest in another.
+        let mut bytes = Vec::new();
+        for _ in 0..3 {
+            bytes.extend_from_slice(&KEEPALIVE);
+        }
+        peer.write_all(&bytes[..47]).await.unwrap();
+        assert!(matches!(frames.next(&ctx).await, Ok(Message::KeepAlive)));
+        assert!(matches!(frames.next(&ctx).await, Ok(Message::KeepAlive)));
+        peer.write_all(&bytes[47..]).await.unwrap();
+        assert!(matches!(frames.next(&ctx).await, Ok(Message::KeepAlive)));
+
+        drop(peer);
+        assert!(matches!(
+            frames.next(&ctx).await,
+            Err(SessionError::Io(e)) if e.kind() == io::ErrorKind::UnexpectedEof
+        ));
+    }
+
+    /// The old reader used read_exact inside select!: when a timer fired
+    /// while a message was half read, the half was thrown away and the
+    /// stream was out of step from then on.
+    #[tokio::test]
+    async fn a_read_abandoned_half_way_loses_nothing() {
+        let (mut peer, ours) = duplex(4096);
+        let mut frames = FrameReader::new(ours);
+        let ctx = ParserContext::default();
+
+        peer.write_all(&KEEPALIVE[..10]).await.unwrap();
+        // Something else wins the race, as a timer would in the session.
+        let abandoned = tokio::time::timeout(Duration::from_millis(20), frames.next(&ctx)).await;
+        assert!(abandoned.is_err(), "half a message is not a message");
+
+        peer.write_all(&KEEPALIVE[10..]).await.unwrap();
+        assert!(matches!(frames.next(&ctx).await, Ok(Message::KeepAlive)));
+    }
+
+    #[tokio::test]
+    async fn an_impossible_length_is_rejected_from_the_header_alone() {
+        let (mut peer, ours) = duplex(4096);
+        let mut frames = FrameReader::new(ours);
+
+        let mut header = KEEPALIVE;
+        header[16..18].copy_from_slice(&5000u16.to_be_bytes());
+        peer.write_all(&header).await.unwrap();
+
+        // No waiting for 5000 bytes that should never be sent.
+        match frames.next(&ParserContext::default()).await {
+            Err(SessionError::Decode(e)) => {
+                assert_eq!(
+                    e.notification,
+                    BgpError::MsgHeader(MsgHeaderSubcode::BadMessageLength)
+                );
+                assert_eq!(e.data, 5000u16.to_be_bytes());
+            }
+            other => panic!("expected a decode error, got {other:?}"),
+        }
+    }
+}

crates/server/src/dialer.rs

@@ -17,7 +17,7 @@ use tracing::{info, warn};
 
 use crate::collision::Direction;
 use crate::config::PeerConfig;
-use crate::fsm::{SessionConfig, run_session};
+use crate::connection::{SessionConfig, run_session};
 use crate::session::{DialState, SessionRegistry};
 
 /// Upper bound for exponential backoff between connection attempts.

crates/server/src/error.rs

@@ -8,23 +8,7 @@ use bgp_packet::message::NotificationMessage;
 use bgp_primitives::parser::EncodeError;
 use thiserror::Error;
 
-/// The states of a session that has a TCP connection (RFC 4271 §8.2.2).
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum SessionState {
-    OpenSent,
-    OpenConfirm,
-    Established,
-}
-
-impl std::fmt::Display for SessionState {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(match self {
-            Self::OpenSent => "OpenSent",
-            Self::OpenConfirm => "OpenConfirm",
-            Self::Established => "Established",
-        })
-    }
-}
+use crate::fsm::State;
 
 /// Why a session ended abnormally.
 #[derive(Debug, Error)]
@@ -37,14 +21,17 @@ pub enum SessionError {
     Encode(#[from] EncodeError),
     /// The peer sent nothing for as long as the pre-negotiation hold time.
     #[error("timed out waiting for a message in the {0} state")]
-    Timeout(SessionState),
+    Timeout(State),
     #[error("hold timer expired")]
     HoldTimerExpired,
     #[error("received {received} in the {state} state")]
     UnexpectedMessage {
-        state: SessionState,
+        state: State,
         received: &'static str,
     },
+    /// The peer ended the session. Nothing is sent in return.
+    #[error("peer sent NOTIFICATION: {0}")]
+    NotifiedByPeer(BgpError),
     #[error("refused the peer's OPEN: {0}")]
     OpenRefused(BgpError),
     #[error("peer announced AS{actual} but AS{expected} was configured")]
@@ -58,7 +45,7 @@ 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(_) => return None,
+            Self::Io(_) | Self::Encode(_) | Self::NotifiedByPeer(_) => return None,
             Self::Decode(e) => {
                 return Some(NotificationMessage {
                     error: e.notification.clone(),
@@ -67,9 +54,11 @@ impl SessionError {
             }
             Self::Timeout(_) | Self::HoldTimerExpired => BgpError::HoldTimer,
             Self::UnexpectedMessage { state, .. } => BgpError::FsmError(match state {
-                SessionState::OpenSent => FsmSubcode::UnexpectedOpenSent,
-                SessionState::OpenConfirm => FsmSubcode::UnexpectedOpenConfirm,
-                SessionState::Established => FsmSubcode::UnexpectedEstablished,
+                State::OpenSent => FsmSubcode::UnexpectedOpenSent,
+                State::OpenConfirm => FsmSubcode::UnexpectedOpenConfirm,
+                // A closed machine handles no messages; Established is the
+                // only other state one can arrive in.
+                State::Established | State::Closed => FsmSubcode::UnexpectedEstablished,
             }),
             Self::OpenRefused(error) => error.clone(),
             Self::WrongPeerAs { .. } => BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs),
@@ -173,12 +162,12 @@ mod tests {
             ))
         );
         assert_eq!(
-            notification(SessionError::Timeout(SessionState::OpenSent)),
+            notification(SessionError::Timeout(State::OpenSent)),
             Some((BgpError::HoldTimer, vec![]))
         );
         assert_eq!(
             notification(SessionError::UnexpectedMessage {
-                state: SessionState::OpenConfirm,
+                state: State::OpenConfirm,
                 received: "update"
             }),
             Some((

crates/server/src/server.rs

@@ -11,10 +11,9 @@ use bgp_rib::{AdjRibIn, LocalRib};
 use crate::advertise::spawn_advertiser;
 use crate::collision::{CollisionTable, Direction};
 use crate::config::Config;
-use crate::connection::handle_connection;
+use crate::connection::{SessionConfig, handle_connection};
 use crate::dialer::spawn_dialers;
 use crate::error::ServerError;
-use crate::fsm::SessionConfig;
 use crate::session::{ConfiguredPeer, SessionRegistry};
 
 pub struct BgpServerBuilder {