feature: RFC 4271 §6.8 connection collision detection

When both ends dial each other, exactly one TCP connection must survive
and both ends must pick the same one. Previously ferrix relied on the
other router to sort this out and could briefly hold two sessions.

- collision: a shared map keyed by the peer's BGP Identifier holds the one
  connection currently preferred for that peer plus a kill signal. Each
  connection calls admit() on receiving the peer's OPEN. The connection
  initiated by the higher BGP Identifier wins; the loser gets Cease /
  Connection Collision Resolution, whether it is the new connection or an
  existing one in OpenConfirm running in another task.
- A collision with an Established session closes the new connection (RFC
  default). The optional CollisionDetectEstablishedState is available as
  collision_detect_established_state: a restarted peer's new connection
  then replaces the stale session at once instead of being refused until
  the hold timer expires. Enabled in the lab config.
- Two connections opened by the same side: the newer wins.
- OPEN validation rejects a peer using our own BGP Identifier or 0.0.0.0
  with Bad BGP Identifier; collision resolution needs distinct identifiers.
- OpenConfirm no longer times out instantly when the negotiated hold time
  is zero.
- Sessions record which side opened them; shown as "opened by" on the
  status page and as `direction` in /api/peers. New counter
  bgp_connection_collisions_total{closed}.
- Tests: unit tests for the rule (including that both ends agree), and
  integration tests for simultaneous open in both identifier orders and
  both arrival orders, Established default, restart replacement with route
  flush, distinct identifiers, slot release, and Bad BGP Identifier.
  Existing fan-out test now gives its two peers distinct identifiers.

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

crates/integration-tests/src/test_server.rs

@@ -58,6 +58,7 @@ impl TestServer {
             listen_endpoints: vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)],
             peers: vec![],
             connect_retry: std::time::Duration::from_secs(5),
+            collision_detect_established_state: false,
             local_route_policy: LocalRibPolicy::default(),
             http_endpoint: None,
         };

crates/integration-tests/tests/announce.rs

@@ -28,8 +28,14 @@ fn route(p: &str, attrs: LocalRouteAttrs) -> LocalRoute {
 }
 
 async fn established_peer(server: &TestServer) -> Result<TestPeer> {
+    established_peer_with_id(server, Ipv4Addr::new(10, 0, 0, 1)).await
+}
+
+/// Distinct routers need distinct BGP Identifiers, or the second connection
+/// is (correctly) treated as a collision with the first.
+async fn established_peer_with_id(server: &TestServer, router_id: Ipv4Addr) -> Result<TestPeer> {
     let mut peer = TestPeer::connect(server.addr).await?;
-    peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 90).await?;
+    peer.handshake(PEER_ASN, router_id, 90).await?;
     Ok(peer)
 }
 
@@ -143,8 +149,8 @@ async fn test_routes_replayed_on_session_up() -> Result<()> {
 #[tokio::test]
 async fn test_announce_reaches_all_sessions() -> Result<()> {
     let server = TestServer::start(LOCAL_ASN).await?;
-    let mut a = established_peer(&server).await?;
+    let mut a = established_peer_with_id(&server, Ipv4Addr::new(10, 0, 0, 1)).await?;
-    let mut b = established_peer(&server).await?;
+    let mut b = established_peer_with_id(&server, Ipv4Addr::new(10, 0, 0, 2)).await?;
     wait_for_sessions(&server, 2).await?;
 
     server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?;

crates/integration-tests/tests/collision.rs

@@ -0,0 +1,287 @@
+//! Connection collision detection (RFC 4271 §6.8) against a live server.
+//!
+//! The test server's BGP Identifier is 127.0.0.1. A test peer picks an
+//! identifier above or below that to steer which connection must survive.
+
+use std::net::{Ipv4Addr, SocketAddr};
+use std::time::Duration;
+
+use bgp_integration_tests::test_peer::TestPeer;
+use bgp_integration_tests::test_server::TestServer;
+use bgp_packet::errors::{BgpError, CeaseSubcode, OpenMsgSubcode};
+use bgp_packet::message::{
+    AsPathAttribute, Message, NextHopPathAttribute, NotificationMessage, OriginPathAttribute,
+    PathAttribute, UpdateMessage,
+};
+use bgp_primitives::ip_prefix::IpPrefix;
+use bgp_rib::{LocalRoute, LocalRouteAttrs};
+use bgp_server::collision::Direction;
+use bgp_server::config::PeerConfig;
+use eyre::Result;
+use tokio::net::TcpListener;
+
+const LOCAL_ASN: u32 = 65000;
+const PEER_ASN: u32 = 65001;
+const TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Above the server's 127.0.0.1.
+const HIGH_ID: Ipv4Addr = Ipv4Addr::new(192, 0, 2, 1);
+/// Below the server's 127.0.0.1.
+const LOW_ID: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 1);
+
+async fn server_dialing(listener: &TcpListener) -> Result<TestServer> {
+    let endpoint: SocketAddr = listener.local_addr()?;
+    TestServer::start_with(LOCAL_ASN, |c| {
+        c.connect_retry = Duration::from_millis(200);
+        c.peers = vec![PeerConfig { name: "lab".into(), endpoint, remote_asn: PEER_ASN, hold_time_secs: 90 }];
+    })
+    .await
+}
+
+/// Exchanges OPENs only, leaving the server's side in OpenConfirm (it has
+/// sent its KEEPALIVE and is waiting for ours).
+async fn open_only(peer: &mut TestPeer, router_id: Ipv4Addr) -> Result<()> {
+    peer.send_open(PEER_ASN, router_id, 90).await?;
+    peer.recv_open().await?;
+    Ok(())
+}
+
+/// Finishes the handshake begun by [`open_only`].
+async fn confirm(peer: &mut TestPeer) -> Result<()> {
+    match tokio::time::timeout(TIMEOUT, peer.recv()).await?? {
+        Message::KeepAlive => {}
+        other => eyre::bail!("expected KEEPALIVE, got {other:?}"),
+    }
+    peer.send(&Message::KeepAlive).await
+}
+
+/// Reads until a NOTIFICATION arrives, skipping KEEPALIVEs.
+async fn expect_notification(peer: &mut TestPeer) -> Result<NotificationMessage> {
+    tokio::time::timeout(TIMEOUT, async {
+        loop {
+            match peer.recv().await? {
+                Message::Notification(n) => return Ok(n),
+                Message::KeepAlive => {}
+                other => eyre::bail!("expected NOTIFICATION, got {other:?}"),
+            }
+        }
+    })
+    .await?
+}
+
+fn is_collision_cease(n: &NotificationMessage) -> bool {
+    matches!(n.error, BgpError::Cease(CeaseSubcode::ConnCollisionResolution))
+}
+
+async fn wait_for_sessions(server: &TestServer, n: usize) -> Result<()> {
+    tokio::time::timeout(TIMEOUT, async {
+        while server.registry.list().len() != n {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await?;
+    Ok(())
+}
+
+/// Peer has the higher identifier, so the connection the *peer* initiated
+/// must survive: the server closes its own outbound connection.
+#[tokio::test]
+async fn test_simultaneous_open_peer_higher_keeps_inbound() -> Result<()> {
+    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
+    let server = server_dialing(&listener).await?;
+
+    // Server's outbound connection reaches OpenConfirm.
+    let (stream, _) = tokio::time::timeout(TIMEOUT, listener.accept()).await??;
+    let mut dialled = TestPeer::from_stream(stream);
+    open_only(&mut dialled, HIGH_ID).await?;
+
+    // The peer now also dials the server.
+    let mut inbound = TestPeer::connect(server.addr).await?;
+    open_only(&mut inbound, HIGH_ID).await?;
+
+    let notif = expect_notification(&mut dialled).await?;
+    assert!(is_collision_cease(&notif), "{notif:?}");
+    assert!(dialled.is_closed().await);
+
+    confirm(&mut inbound).await?;
+    wait_for_sessions(&server, 1).await?;
+    let session = server.registry.list().remove(0);
+    assert_eq!(session.direction, Direction::Inbound);
+    assert_eq!(session.name.as_deref(), Some("lab"));
+
+    // The survivor is fully functional.
+    server
+        .local_rib
+        .announce(LocalRoute::new(IpPrefix::try_from("10.100.1.0/24")?, "alice", LocalRouteAttrs::default()))
+        .await?;
+    let update = tokio::time::timeout(TIMEOUT, inbound.recv_update()).await??;
+    assert_eq!(update.announced_routes[0].to_string(), "10.100.1.0/24");
+
+    // The dialer notices the surviving session and does not open another.
+    let extra = tokio::time::timeout(Duration::from_millis(700), listener.accept()).await;
+    assert!(extra.is_err(), "dialer opened a second connection despite an established session");
+    Ok(())
+}
+
+/// Peer has the lower identifier, so the connection the *server* initiated
+/// must survive: the peer's inbound connection is refused.
+#[tokio::test]
+async fn test_simultaneous_open_peer_lower_keeps_outbound() -> Result<()> {
+    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
+    let server = server_dialing(&listener).await?;
+
+    let (stream, _) = tokio::time::timeout(TIMEOUT, listener.accept()).await??;
+    let mut dialled = TestPeer::from_stream(stream);
+    open_only(&mut dialled, LOW_ID).await?;
+
+    let mut inbound = TestPeer::connect(server.addr).await?;
+    open_only(&mut inbound, LOW_ID).await?;
+
+    let notif = expect_notification(&mut inbound).await?;
+    assert!(is_collision_cease(&notif), "{notif:?}");
+    assert!(inbound.is_closed().await);
+
+    confirm(&mut dialled).await?;
+    wait_for_sessions(&server, 1).await?;
+    assert_eq!(server.registry.list()[0].direction, Direction::Outbound);
+    Ok(())
+}
+
+/// The order in which the two OPENs are processed must not change the
+/// outcome: here the inbound connection reaches OpenConfirm first.
+#[tokio::test]
+async fn test_outcome_is_independent_of_arrival_order() -> Result<()> {
+    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
+    let server = server_dialing(&listener).await?;
+
+    let (stream, _) = tokio::time::timeout(TIMEOUT, listener.accept()).await??;
+    let mut dialled = TestPeer::from_stream(stream);
+
+    // Inbound first this time; peer is lower so outbound must still win.
+    let mut inbound = TestPeer::connect(server.addr).await?;
+    open_only(&mut inbound, LOW_ID).await?;
+    open_only(&mut dialled, LOW_ID).await?;
+
+    let notif = expect_notification(&mut inbound).await?;
+    assert!(is_collision_cease(&notif), "{notif:?}");
+
+    confirm(&mut dialled).await?;
+    wait_for_sessions(&server, 1).await?;
+    assert_eq!(server.registry.list()[0].direction, Direction::Outbound);
+    Ok(())
+}
+
+/// RFC default: a collision with an Established session closes the new
+/// connection, whatever the identifiers say.
+#[tokio::test]
+async fn test_established_session_wins_by_default() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+
+    let mut first = TestPeer::connect(server.addr).await?;
+    first.handshake(PEER_ASN, HIGH_ID, 90).await?;
+    wait_for_sessions(&server, 1).await?;
+
+    let mut second = TestPeer::connect(server.addr).await?;
+    open_only(&mut second, HIGH_ID).await?;
+    let notif = expect_notification(&mut second).await?;
+    assert!(is_collision_cease(&notif), "{notif:?}");
+
+    // The original is untouched.
+    assert_eq!(server.registry.list().len(), 1);
+    server
+        .local_rib
+        .announce(LocalRoute::new(IpPrefix::try_from("10.100.1.0/24")?, "alice", LocalRouteAttrs::default()))
+        .await?;
+    let update = tokio::time::timeout(TIMEOUT, first.recv_update()).await??;
+    assert_eq!(update.announced_routes[0].to_string(), "10.100.1.0/24");
+    Ok(())
+}
+
+/// With collision_detect_established_state, a peer that restarts and dials
+/// again replaces the stale session at once, and the routes learned over the
+/// old session are flushed.
+#[tokio::test]
+async fn test_restarted_peer_replaces_stale_session_when_enabled() -> Result<()> {
+    let server = TestServer::start_with(LOCAL_ASN, |c| c.collision_detect_established_state = true).await?;
+
+    let mut stale = TestPeer::connect(server.addr).await?;
+    stale.handshake(PEER_ASN, HIGH_ID, 90).await?;
+    wait_for_sessions(&server, 1).await?;
+    stale
+        .send(&Message::Update(UpdateMessage {
+            withdrawn_routes: vec![],
+            path_attributes: vec![
+                PathAttribute::Origin(OriginPathAttribute::IGP),
+                AsPathAttribute::from_asns(vec![PEER_ASN]),
+                PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(10, 0, 0, 9))),
+            ],
+            announced_routes: vec![IpPrefix::try_from("203.0.113.0/24")?],
+        }))
+        .await?;
+    tokio::time::timeout(TIMEOUT, async {
+        while server.adj_rib_in.snapshot().await.is_empty() {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await?;
+    let stale_addr = server.registry.list()[0].peer.addr;
+
+    // "Restart": a fresh connection from the same router.
+    let mut fresh = TestPeer::connect(server.addr).await?;
+    fresh.handshake(PEER_ASN, HIGH_ID, 90).await?;
+
+    let notif = expect_notification(&mut stale).await?;
+    assert!(is_collision_cease(&notif), "{notif:?}");
+
+    tokio::time::timeout(TIMEOUT, async {
+        loop {
+            let sessions = server.registry.list();
+            if sessions.len() == 1 && sessions[0].peer.addr != stale_addr {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await?;
+    assert!(server.adj_rib_in.snapshot().await.is_empty(), "stale routes were not flushed");
+    Ok(())
+}
+
+/// Different routers (different identifiers) never collide.
+#[tokio::test]
+async fn test_distinct_identifiers_do_not_collide() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    let mut a = TestPeer::connect(server.addr).await?;
+    a.handshake(PEER_ASN, HIGH_ID, 90).await?;
+    let mut b = TestPeer::connect(server.addr).await?;
+    b.handshake(PEER_ASN, LOW_ID, 90).await?;
+    wait_for_sessions(&server, 2).await?;
+    Ok(())
+}
+
+/// A failed connection frees its slot: the same router can come back.
+#[tokio::test]
+async fn test_identifier_is_released_when_connection_ends() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    let mut first = TestPeer::connect(server.addr).await?;
+    first.handshake(PEER_ASN, HIGH_ID, 90).await?;
+    wait_for_sessions(&server, 1).await?;
+    drop(first);
+    wait_for_sessions(&server, 0).await?;
+
+    let mut second = TestPeer::connect(server.addr).await?;
+    second.handshake(PEER_ASN, HIGH_ID, 90).await?;
+    wait_for_sessions(&server, 1).await?;
+    Ok(())
+}
+
+/// A peer claiming our own identifier is refused outright.
+#[tokio::test]
+async fn test_peer_with_our_identifier_is_refused() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    let mut peer = TestPeer::connect(server.addr).await?;
+    open_only(&mut peer, Ipv4Addr::new(127, 0, 0, 1)).await?;
+    let notif = expect_notification(&mut peer).await?;
+    assert!(matches!(notif.error, BgpError::OpenMsg(OpenMsgSubcode::BadBgpId)), "{notif:?}");
+    Ok(())
+}

crates/server/proto/config.proto

@@ -28,6 +28,15 @@ message BgpConfig {
   // Initial delay between attempts to connect to a configured peer, in
   // seconds. Doubles after each failure up to two minutes. Defaults to 5.
   uint32 connect_retry_seconds = 7;
+
+  // Connection collisions (RFC 4271 §6.8) against a session that is already
+  // Established are, by default, resolved by closing the new connection.
+  // When true they are resolved like any other collision: the connection
+  // initiated by the higher BGP Identifier survives, and a second connection
+  // opened by the same side replaces the first. This lets a peer that
+  // restarted take over from a session we have not yet noticed is dead,
+  // instead of being refused until the hold timer expires.
+  bool collision_detect_established_state = 8;
 }
 
 // Limits on locally originated routes.

crates/server/src/collision.rs

@@ -0,0 +1,215 @@
+//! Connection collision detection (RFC 4271 §6.8).
+//!
+//! Two speakers that dial each other can end up with two TCP connections.
+//! Exactly one must survive and both ends must pick the same one: the
+//! connection *initiated by the speaker with the higher BGP Identifier*.
+//! The other is closed with Cease / Connection Collision Resolution.
+//!
+//! 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.
+//!
+//! A collision with an *Established* connection closes the new one, unless
+//! `detect_established` is set (the RFC's optional
+//! `CollisionDetectEstablishedState`), which lets a restarted peer replace a
+//! session we have not yet noticed is dead. If both connections were opened
+//! by the same side the newer wins: the initiator has abandoned the older.
+
+use std::collections::HashMap;
+use std::net::Ipv4Addr;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
+
+use tokio::sync::oneshot;
+
+use crate::metrics::COLLISIONS;
+
+/// Which side opened the TCP connection.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Direction {
+    /// The peer connected to us.
+    Inbound,
+    /// We connected to the peer.
+    Outbound,
+}
+
+impl Direction {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Direction::Inbound => "inbound",
+            Direction::Outbound => "outbound",
+        }
+    }
+}
+
+struct Entry {
+    id: u64,
+    direction: Direction,
+    established: bool,
+    kill: Option<oneshot::Sender<()>>,
+}
+
+/// True if the new connection is the one to close.
+fn new_loses(
+    local_id: Ipv4Addr,
+    remote_id: Ipv4Addr,
+    existing: &Entry,
+    new: Direction,
+    detect_established: bool,
+) -> bool {
+    if existing.established && !detect_established {
+        return true;
+    }
+    if existing.direction == new {
+        return false;
+    }
+    // Identifiers compare as unsigned 32-bit integers.
+    let keep = if u32::from(local_id) > u32::from(remote_id) { Direction::Outbound } else { Direction::Inbound };
+    new != keep
+}
+
+/// Shared by every connection of one server.
+#[derive(Clone)]
+pub struct CollisionTable {
+    entries: Arc<Mutex<HashMap<Ipv4Addr, Entry>>>,
+    next_id: Arc<AtomicU64>,
+    detect_established: bool,
+}
+
+impl CollisionTable {
+    pub fn new(detect_established: bool) -> Self {
+        Self { entries: Default::default(), next_id: Default::default(), detect_established }
+    }
+
+    /// Call when a connection receives the peer's OPEN. `None` means this
+    /// connection lost and must be closed with Cease / Connection Collision
+    /// Resolution. Otherwise any rival has been told to close and this
+    /// connection holds the slot until the guard drops.
+    pub fn admit(&self, local_id: Ipv4Addr, remote_id: Ipv4Addr, direction: Direction) -> Option<ConnectionGuard> {
+        let mut entries = self.entries.lock().expect("collision table poisoned");
+        if let Some(existing) = entries.get_mut(&remote_id) {
+            if new_loses(local_id, remote_id, existing, direction, self.detect_established) {
+                COLLISIONS.with_label_values(&["new"]).inc();
+                return None;
+            }
+            if let Some(kill) = existing.kill.take() {
+                COLLISIONS.with_label_values(&["existing"]).inc();
+                let _ = kill.send(());
+            }
+        }
+        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
+        let (kill, killed) = oneshot::channel();
+        entries.insert(remote_id, Entry { id, direction, established: false, kill: Some(kill) });
+        Some(ConnectionGuard { table: self.clone(), remote_id, id, killed })
+    }
+}
+
+/// Holds a connection's slot in the table; releases it on drop.
+pub struct ConnectionGuard {
+    table: CollisionTable,
+    remote_id: Ipv4Addr,
+    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<()>,
+}
+
+impl ConnectionGuard {
+    pub fn set_established(&self) {
+        let mut entries = self.table.entries.lock().expect("collision table poisoned");
+        if let Some(e) = entries.get_mut(&self.remote_id).filter(|e| e.id == self.id) {
+            e.established = true;
+        }
+    }
+}
+
+impl Drop for ConnectionGuard {
+    fn drop(&mut self) {
+        // 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.lock().expect("collision table poisoned");
+        if entries.get(&self.remote_id).is_some_and(|e| e.id == self.id) {
+            entries.remove(&self.remote_id);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use Direction::*;
+
+    const LOW: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 1);
+    const HIGH: Ipv4Addr = Ipv4Addr::new(192, 0, 2, 1);
+
+    fn loses(local: Ipv4Addr, remote: Ipv4Addr, existing: Direction, established: bool, new: Direction, detect: bool) -> bool {
+        let e = Entry { id: 0, direction: existing, established, kill: None };
+        new_loses(local, remote, &e, new, detect)
+    }
+
+    #[test]
+    fn higher_identifier_keeps_the_connection_it_initiated() {
+        // We are HIGH: our outbound connection survives.
+        assert!(loses(HIGH, LOW, Outbound, false, Inbound, false));
+        assert!(!loses(HIGH, LOW, Inbound, false, Outbound, false));
+        // We are LOW: the peer's connection (inbound to us) survives.
+        assert!(!loses(LOW, HIGH, Outbound, false, Inbound, false));
+        assert!(loses(LOW, HIGH, Inbound, false, Outbound, false));
+    }
+
+    #[test]
+    fn both_ends_agree() {
+        // A and B dial each other. Each end holds its own dial as the
+        // existing (Outbound) connection and sees the other's dial arrive as
+        // new (Inbound). Exactly one of the two dials may be closed, so the
+        // ends must give opposite answers to "close the other side's dial?".
+        for (a, b) in [(LOW, HIGH), (HIGH, LOW)] {
+            let a_closes_dial_from_b = loses(a, b, Outbound, false, Inbound, false);
+            let b_closes_dial_from_a = loses(b, a, Outbound, false, Inbound, false);
+            assert_ne!(a_closes_dial_from_b, b_closes_dial_from_a, "ends disagree for a={a} b={b}");
+        }
+    }
+
+    #[test]
+    fn established_wins_unless_enabled() {
+        for new in [Inbound, Outbound] {
+            for existing in [Inbound, Outbound] {
+                assert!(loses(LOW, HIGH, existing, true, new, false));
+                assert!(loses(HIGH, LOW, existing, true, new, false));
+            }
+        }
+        assert!(!loses(LOW, HIGH, Outbound, true, Inbound, true));
+        assert!(loses(HIGH, LOW, Outbound, true, Inbound, true));
+    }
+
+    #[test]
+    fn same_direction_newer_wins() {
+        assert!(!loses(LOW, HIGH, Inbound, false, Inbound, false));
+        assert!(!loses(HIGH, LOW, Inbound, false, Inbound, false));
+        assert!(loses(LOW, HIGH, Inbound, true, Inbound, false));
+        assert!(!loses(LOW, HIGH, Inbound, true, Inbound, true));
+    }
+
+    #[tokio::test]
+    async fn table_signals_loser_and_releases_slots() {
+        let table = CollisionTable::new(false);
+        // 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();
+        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());
+
+        // A different identifier never collides.
+        assert!(table.admit(LOW, 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());
+    }
+}

crates/server/src/config.rs

@@ -23,6 +23,10 @@ pub struct Config {
     /// Initial delay between attempts to connect to a configured peer;
     /// doubles on each failure.
     pub connect_retry: Duration,
+    /// Resolve collisions against Established sessions by BGP Identifier
+    /// instead of always closing the new connection (RFC 4271 §8.1.1
+    /// `CollisionDetectEstablishedState`).
+    pub collision_detect_established_state: bool,
     /// Limits on locally originated routes.
     pub local_route_policy: LocalRibPolicy,
     /// Address for the HTTP API; `None` disables it.
@@ -121,6 +125,7 @@ impl Config {
             listen_endpoints,
             peers,
             connect_retry,
+            collision_detect_established_state: proto.collision_detect_established_state,
             local_route_policy,
             http_endpoint,
         })

crates/server/src/dialer.rs

@@ -3,9 +3,10 @@
 //! One task per configured peer loops forever: connect, run the session,
 //! back off, repeat. While a session with the same remote IP and ASN is
 //! already established (for example because the peer dialled us first) the
-//! dialer stays idle rather than opening a second one. Full RFC 4271 §6.8
+//! dialer stays idle rather than opening a second one. When both sides dial
-//! collision detection is not implemented; a real router on the other end
+//! at once the FSM resolves the collision (see [`crate::collision`]); if our
-//! will resolve any remaining race by closing one of the two connections.
+//! connection is the one closed, the next pass finds the surviving session
+//! and goes idle.
 
 use std::sync::Arc;
 use std::time::{Duration, SystemTime};
@@ -14,6 +15,7 @@ use tokio::net::TcpStream;
 use tokio::task::JoinHandle;
 use tracing::{info, warn};
 
+use crate::collision::Direction;
 use crate::config::PeerConfig;
 use crate::fsm::{SessionConfig, run_session};
 use crate::session::{DialState, SessionRegistry};
@@ -32,6 +34,7 @@ pub fn spawn_dialers(
         .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),
                 ..base.clone()

crates/server/src/fsm.rs

@@ -1,7 +1,7 @@
 use std::net::{Ipv4Addr, SocketAddr};
 use std::time::Duration;
 
-use bgp_packet::errors::{BgpError, FsmSubcode, OpenMsgSubcode};
+use bgp_packet::errors::{BgpError, CeaseSubcode, FsmSubcode, OpenMsgSubcode};
 use bgp_packet::message::{BGP_HEADER_LEN, Message, NotificationMessage, PathAttribute};
 use bgp_packet::open::{Capability, OpenMessage, OpenOption};
 use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi};
@@ -12,12 +12,13 @@ use eyre::{Result, bail, eyre};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
 use tokio::time::{Instant, interval_at, sleep_until};
 use tracing::{info, warn};
 
 use bgp_packet::message::{AS_TRANS, BGP4_VERSION};
 
+use crate::collision::{CollisionTable, Direction};
 use crate::metrics::{MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, message_type_name};
 use crate::session::{SessionHandle, SessionRegistry};
 
@@ -33,6 +34,10 @@ pub struct SessionConfig {
     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
@@ -87,7 +92,7 @@ pub async fn run_session(
         }
     };
 
-    if let Err(bgp_err) = validate_open(&peer_open) {
+    if let Err(bgp_err) = validate_open(&peer_open, config.router_id) {
         send_notification(&mut writer, bgp_err, &ParserContext::default()).await;
         bail!("Peer {} sent invalid OPEN", peer_addr);
     }
@@ -124,19 +129,39 @@ pub async fn run_session(
         bail!("Peer {peer_addr} announced AS{peer_asn} but AS{expected} was configured");
     }
 
+    // Connection collision detection (RFC 4271 §6.8). From here on the
+    // guard keeps this connection visible to later ones, and `killed` fires
+    // if one of them wins.
+    let Some(mut collision_guard) =
+        config.collisions.admit(config.router_id, peer_open.identifier, config.direction)
+    else {
+        return close_for_collision(&mut writer, &ctx, peer_addr).await;
+    };
+
     info!(
         %peer_addr,
         peer_asn,
         negotiated_hold_time,
+        direction = config.direction.as_str(),
         "OPEN received, transitioning to OpenConfirm"
     );
 
     // --- OpenConfirm --------------------------------------------------------
     send_message(&mut writer, &Message::KeepAlive, &ctx).await?;
 
-    let hold_duration = Duration::from_secs(negotiated_hold_time as u64);
+    // With a negotiated hold time of zero the hold timer does not run
+    // (RFC 4271 §4.2); fall back to the large pre-negotiation value so a
+    // silent peer is still cleaned up eventually.
+    let hold_duration = Duration::from_secs(if negotiated_hold_time == 0 { 240 } else { negotiated_hold_time as u64 });
 
-    match recv_with_timeout(&mut reader, &ctx, hold_duration).await {
+    let confirm = tokio::select! {
+        msg = recv_with_timeout(&mut reader, &ctx, hold_duration) => msg,
+        _ = &mut collision_guard.killed => {
+            return close_for_collision(&mut writer, &ctx, peer_addr).await;
+        }
+    };
+
+    match confirm {
         Ok(Message::KeepAlive) => {}
         Ok(other) => {
             send_notification(
@@ -162,9 +187,11 @@ pub async fn run_session(
     // Register so the rest of the server can see this session and queue
     // messages for it. The guard deregisters on any exit path.
     let (outbound_tx, mut outbound_rx) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY);
+    collision_guard.set_established();
     let _guard = config.registry.register(SessionHandle::new(
         peer_id,
         config.peer_name.clone(),
+        config.direction,
         local_addr,
         peer_four_byte_asn,
         outbound_tx,
@@ -177,7 +204,7 @@ pub async fn run_session(
         negotiated_hold_time,
         peer_id,
         &config.adj_rib_in,
-        &mut outbound_rx,
+        SessionInputs { outbound: &mut outbound_rx, killed: &mut collision_guard.killed },
     )
     .await;
 
@@ -189,6 +216,14 @@ pub async fn run_session(
     result
 }
 
+/// Signals from the rest of the server into an established session.
+struct SessionInputs<'a> {
+    /// Messages queued for transmission to the peer.
+    outbound: &'a mut mpsc::Receiver<Message>,
+    /// Fires when a later connection won collision resolution.
+    killed: &'a mut oneshot::Receiver<()>,
+}
+
 async fn established_loop(
     reader: &mut OwnedReadHalf,
     writer: &mut OwnedWriteHalf,
@@ -196,8 +231,9 @@ async fn established_loop(
     hold_time_secs: u16,
     peer_id: PeerId,
     adj_rib_in: &AdjRibIn,
-    outbound: &mut mpsc::Receiver<Message>,
+    inputs: SessionInputs<'_>,
 ) -> Result<()> {
+    let SessionInputs { outbound, killed } = inputs;
     // When hold_time is 0, keepalives and the hold timer are disabled
     // (RFC 4271 §4.4). The timer branches below are gated off in that case;
     // the placeholder durations are never observed.
@@ -255,6 +291,9 @@ async fn established_loop(
                 }
                 send_message(writer, &msg, ctx).await?;
             }
+            _ = &mut *killed => {
+                return close_for_collision(writer, ctx, peer_id.addr).await;
+            }
             _ = keepalive_timer.tick(), if timers_enabled => {
                 send_message(writer, &Message::KeepAlive, ctx).await?;
             }
@@ -326,6 +365,18 @@ async fn send_notification(writer: &mut OwnedWriteHalf, error: BgpError, ctx: &P
     }
 }
 
+/// A later connection to the same BGP Identifier won collision resolution:
+/// tell the peer why and end this session.
+async fn close_for_collision(
+    writer: &mut OwnedWriteHalf,
+    ctx: &ParserContext,
+    peer_addr: SocketAddr,
+) -> Result<()> {
+    info!(%peer_addr, "Closing connection: lost collision resolution");
+    send_notification(writer, BgpError::Cease(CeaseSubcode::ConnCollisionResolution), ctx).await;
+    bail!("Connection to {peer_addr} closed by collision resolution");
+}
+
 // ---------------------------------------------------------------------------
 // RIB helpers
 // ---------------------------------------------------------------------------
@@ -398,10 +449,15 @@ fn build_open(config: &SessionConfig) -> Message {
     })
 }
 
-fn validate_open(open: &OpenMessage) -> Result<(), BgpError> {
+fn validate_open(open: &OpenMessage, local_id: Ipv4Addr) -> Result<(), BgpError> {
     if open.version != BGP4_VERSION {
         return Err(BgpError::OpenMsg(OpenMsgSubcode::UnsupportedVersion));
     }
+    // A peer cannot share our identifier or use the unspecified address;
+    // collision resolution depends on identifiers being distinct.
+    if open.identifier == local_id || open.identifier.is_unspecified() {
+        return Err(BgpError::OpenMsg(OpenMsgSubcode::BadBgpId));
+    }
     if open.hold_time != 0 && open.hold_time < 3 {
         return Err(BgpError::OpenMsg(OpenMsgSubcode::UnacceptableHoldTime));
     }

crates/server/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod advertise;
+pub mod collision;
 pub mod config;
 pub mod connection;
 pub mod dialer;

crates/server/src/metrics.rs

@@ -28,6 +28,15 @@ pub static SESSIONS_ESTABLISHED: LazyLock<IntGauge> = LazyLock::new(|| {
         .expect("metric registration")
 });
 
+pub static COLLISIONS: LazyLock<IntCounterVec> = LazyLock::new(|| {
+    register_int_counter_vec!(
+        "bgp_connection_collisions_total",
+        "Connection collisions resolved, by which connection was closed",
+        &["closed"]
+    )
+    .expect("metric registration")
+});
+
 pub static LOCAL_ROUTES: LazyLock<IntGauge> = LazyLock::new(|| {
     register_int_gauge!("bgp_local_routes", "Locally originated routes").expect("metric registration")
 });
@@ -44,6 +53,7 @@ pub fn init() {
     LazyLock::force(&MESSAGES_RECEIVED);
     LazyLock::force(&SESSIONS_ESTABLISHED_TOTAL);
     LazyLock::force(&SESSIONS_ESTABLISHED);
+    LazyLock::force(&COLLISIONS);
     LazyLock::force(&LOCAL_ROUTES);
     LazyLock::force(&ADJ_RIB_IN_ROUTES);
 }

crates/server/src/server.rs

@@ -10,6 +10,7 @@ use tracing::{error, info};
 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::dialer::spawn_dialers;
@@ -98,6 +99,8 @@ impl BgpServer {
             hold_time_secs: self.config.hold_time_secs,
             adj_rib_in: self.adj_rib_in.clone(),
             registry: self.registry.clone(),
+            collisions: CollisionTable::new(self.config.collision_detect_established_state),
+            direction: Direction::Inbound,
             peer_name: None,
             expected_asn: None,
         };

crates/server/src/session.rs

@@ -14,6 +14,7 @@ use bgp_packet::message::Message;
 use bgp_rib::PeerId;
 use tokio::sync::{broadcast, mpsc};
 
+use crate::collision::Direction;
 use crate::metrics::SESSIONS_ESTABLISHED;
 
 const EVENT_CHANNEL_CAPACITY: usize = 256;
@@ -42,6 +43,8 @@ pub struct SessionHandle {
     pub peer: PeerId,
     /// Name from the configuration, if this peer is configured.
     pub name: Option<String>,
+    /// Which side opened the TCP connection.
+    pub direction: Direction,
     /// Our side of the TCP connection; used as the next hop for routes we
     /// originate to this peer.
     pub local_addr: SocketAddr,
@@ -60,11 +63,12 @@ impl SessionHandle {
     pub fn new(
         peer: PeerId,
         name: Option<String>,
+        direction: Direction,
         local_addr: SocketAddr,
         four_octet_asn: bool,
         outbound: mpsc::Sender<Message>,
     ) -> Self {
-        Self { peer, name, local_addr, established_at: SystemTime::now(), four_octet_asn, outbound }
+        Self { peer, name, direction, local_addr, established_at: SystemTime::now(), four_octet_asn, outbound }
     }
 
     /// Queues a message for transmission on this session. Waits for queue

crates/web/src/api.rs

@@ -83,6 +83,8 @@ pub struct PeerView {
     pub peer_asn: u32,
     /// `Established`, `Connecting`, `Backoff` or `Idle`.
     pub state: String,
+    /// `inbound` or `outbound`: which side opened the live TCP connection.
+    pub direction: Option<String>,
     pub local_addr: Option<String>,
     pub established_at_unix: Option<u64>,
     pub uptime_secs: Option<u64>,
@@ -99,6 +101,7 @@ impl From<&SessionHandle> for PeerView {
             peer_addr: s.peer.addr.to_string(),
             peer_asn: s.peer.asn,
             state: "Established".into(),
+            direction: Some(s.direction.as_str().into()),
             local_addr: Some(s.local_addr.to_string()),
             established_at_unix: Some(unix_secs(s.established_at)),
             uptime_secs: Some(s.established_at.elapsed().map(|d| d.as_secs()).unwrap_or(0)),
@@ -121,6 +124,7 @@ impl PeerView {
             peer_addr: peer.endpoint.to_string(),
             peer_asn: peer.remote_asn,
             state: state_name.into(),
+            direction: None,
             local_addr: None,
             established_at_unix: None,
             uptime_secs: None,

crates/web/static/statusz.html

@@ -82,7 +82,7 @@
 <section id="peers">
   <h2>Peers <span class="count" id="peers-count"></span></h2>
   <table>
-    <thead><tr><th>name</th><th>peer</th><th>ASN</th><th>state</th><th>local address (next hop)</th><th>established</th><th>uptime</th><th>4-byte ASN</th><th>detail</th></tr></thead>
+    <thead><tr><th>name</th><th>peer</th><th>ASN</th><th>state</th><th>opened by</th><th>local address (next hop)</th><th>established</th><th>uptime</th><th>4-byte ASN</th><th>detail</th></tr></thead>
     <tbody id="peers-body"></tbody>
   </table>
 </section>
@@ -170,13 +170,13 @@
   const STATE_CLASS = { Established: 'ok', Connecting: 'warn', Backoff: 'err', Idle: 'muted' };
   function renderPeers() {
     const b = $('peers-body'); $('peers-count').textContent = `${state.peers.filter(p => p.state === 'Established').length} up / ${state.peers.length}`;
-    if (!state.peers.length) { b.innerHTML = '<tr><td colspan="9" class="empty">No peers configured and none connected.</td></tr>'; return; }
+    if (!state.peers.length) { b.innerHTML = '<tr><td colspan="10" class="empty">No peers configured and none connected.</td></tr>'; return; }
     b.innerHTML = state.peers.map(p => {
       const up = p.state === 'Established';
       const detail = p.state === 'Backoff' ? `${esc(p.last_error || '')}; retry in ${fmtDur(p.retry_at_unix - now())}` : '';
       return `<tr>
       <td>${esc(p.name ?? '')}${p.name ? '' : '<span class="muted">(inbound)</span>'}</td><td>${esc(p.peer_addr)}</td><td>${p.peer_asn}</td>
-      <td class="${STATE_CLASS[p.state] || ''}">${esc(p.state)}</td><td>${up ? esc(p.local_addr) : '—'}</td>
+      <td class="${STATE_CLASS[p.state] || ''}">${esc(p.state)}</td><td>${up ? (p.direction === 'outbound' ? 'us' : 'peer') : '—'}</td><td>${up ? esc(p.local_addr) : '—'}</td>
       <td>${up ? fmtTime(p.established_at_unix) : '—'}</td><td>${up ? age(p.established_at_unix) : '—'}</td><td>${up ? (p.four_octet_asn ? 'yes' : 'no') : '—'}</td>
       <td class="wrap muted">${detail}</td></tr>`; }).join('');
   }

docs/RFCs.md

@@ -27,7 +27,9 @@ The base protocol specification.
 | §8 | FSM — passive (inbound) side: OpenSent → OpenConfirm → Established | ✅ |
 | §8 | FSM — active (outbound) side: dial configured peers, exponential retry | ✅ |
 | §6.2 | Bad Peer AS check against configured ASN | ✅ |
-| §6.8 | Connection collision detection (router-ID comparison) | ❌ |
+| §6.8 | Connection collision detection (BGP Identifier comparison, Cease / Connection Collision Resolution) | ✅ |
+| §8.1.1 | Optional `CollisionDetectEstablishedState` (`collision_detect_established_state`) | ✅ |
+| §6.2 | Bad BGP Identifier check (peer using our identifier, or 0.0.0.0) | ✅ |
 | §9 | Decision process / best-path selection | ❌ |
 | §9 | Adj-RIB-In (per-peer table of received routes) | ✅ |
 | §9 | Locally originated routes, advertised to all established peers | ✅ |

docs/lab/README.md

@@ -72,5 +72,11 @@ they have not been executed. Things most likely to need a touch:
   capability there. If you run it on the host instead, either run as root,
   grant `CAP_NET_BIND_SERVICE`, or change `listen_endpoints` and the FRR
   neighbor port.
+- Both ferrix and the FRR routers dial each other, so connection collisions
+  (RFC 4271 §6.8) are normal at start-up: one of the two TCP connections is
+  closed with a Cease / Connection Collision Resolution NOTIFICATION and the
+  other carries the session. ferrix has the lowest BGP Identifier here, so
+  the connection the FRR router opened is the one that survives; the
+  "opened by" column on the status page shows which side that was.
 - ferrix does not check AS paths for its own ASN. That is what lets the
   looped route be accepted; a production BGP speaker would reject it.

docs/lab/ferrix.textpb

@@ -16,6 +16,11 @@ peers {
 }
 connect_retry_seconds: 5
 
+# If r1 or r2 is restarted it will dial us again while we still believe the
+# old session is up. Let that new connection replace the stale one straight
+# away instead of refusing it until the hold timer (30 s here) expires.
+collision_detect_established_state: true
+
 # Students may only announce out of 10.100.0.0/16; one /24 each is plenty.
 local_route_policy {
     allowed_prefixes: "10.100.0.0/16"

testdata/server.textpb

@@ -15,6 +15,9 @@ peers {
     hold_time_seconds: 30
 }
 connect_retry_seconds: 5
+# Let a restarted peer replace a stale Established session immediately
+# (RFC 4271 CollisionDetectEstablishedState). Default false.
+collision_detect_established_state: false
 local_route_policy {
     allowed_prefixes: "10.100.0.0/16"
     allowed_prefixes: "2001:db8:100::/48"