Rework the speaker: sans-I/O FSM, sharded table, RFC 7606, ADD-PATH

The work of the speaker-cleanup branch as one commit; the branch keeps
the step-by-step history.

Sessions
- The state machine is a pure function from events to actions, driven by
  one task per connection. Idle, Connect and Active belong to a dialer
  that keeps configured peers up with backoff.
- Capabilities are negotiated: address families, four-octet AS numbers
  with AS_TRANS and AS4_PATH, route refresh, ADD-PATH per family and
  direction.
- Collision detection per RFC 4271 section 6.8, by address and
  identifier.
- Connections are taken only from configured peers, a few per address.
- A panic in a session is contained and answered with a Cease; a core
  task that stops shuts the server down with a Cease to every peer.
- Writes that make no progress are given up (RFC 9687).

Codec
- Malformed UPDATEs are handled as RFC 7606 says: treat-as-withdraw,
  attribute discard, or session reset, with the NOTIFICATION to send.
- Address families are open types; any can be named and ignored.
- Prefixes are fixed-size values, normalised, with path identifiers.
- Lengths are filled in after writing; one definition of each type code.
- Property tests and fuzz targets for every decoder.

Routing table
- Adj-RIBs-In and Loc-RIB in one sharded table that sessions write to
  directly. Routes are sixteen bytes; attribute sets are interned in
  wire form. Best-path selection per RFC 4271 section 9.1.2.2 with
  deterministic MED.
- A journal per shard lets followers read changes at their own pace;
  one that falls behind is resynchronised.
- On a RIPE RIS dump of 148 peers and 39.3M routes: loaded in 2.0 s on
  12 cores, about 52 bytes a route.

Announcing
- Originated routes come from the configuration. Each session has its
  own exporter with an Adj-RIB-Out, so a slow peer delays only itself.
- Routes to a peer in our own AS have an empty path and a LOCAL_PREF.

Operations
- HTTP status page, JSON API, event stream and Prometheus metrics.
- Benchmarks on synthetic and real tables.

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

crates/server/src/session.rs

@@ -0,0 +1,294 @@
+//! Registry of established BGP sessions.
+//!
+//! A session registers itself when it reaches Established and is removed when
+//! its [`SessionGuard`] drops, so a session that errors out is always
+//! deregistered. Other parts of the server (the advertiser, the HTTP API) use
+//! the registry to enumerate live sessions and queue outbound messages.
+
+use std::collections::HashMap;
+use std::net::{IpAddr, SocketAddr};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, RwLock};
+use std::time::{Duration, SystemTime};
+
+use bgp_packet::errors::{BgpError, CeaseSubcode};
+use bgp_packet::message::NotificationMessage;
+
+use bgp_packet::message::Message;
+use bgp_rib::table::PeerIndex;
+use tokio::sync::{broadcast, mpsc};
+
+use crate::collision::Direction;
+use crate::family::FamilySet;
+use crate::fsm::Negotiated;
+use crate::metrics::SESSIONS_ESTABLISHED;
+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 {
+    /// Not dialling: the peer connected to us, or the dialer has not started.
+    Idle,
+    Connecting,
+    /// The last attempt failed or the session ended; waiting to retry.
+    Backoff {
+        until: SystemTime,
+        /// How the last attempt or session ended.
+        last_error: String,
+    },
+}
+
+/// A peer from the configuration file, whether or not it is currently up.
+#[derive(Debug, Clone)]
+pub struct ConfiguredPeer {
+    pub name: String,
+    pub endpoint: SocketAddr,
+    pub remote_asn: u32,
+}
+
+/// Handle to one established session. Cloning is cheap.
+#[derive(Debug, Clone)]
+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,
+    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,
+    /// The address families negotiated with the peer. Routes of any other
+    /// family are neither sent to it nor accepted from it.
+    pub families: FamilySet,
+    /// The families for which the peer sends us several routes for a
+    /// prefix (ADD-PATH, RFC 7911), and those for which we send it path
+    /// identifiers.
+    pub add_path_receive: FamilySet,
+    pub add_path_send: FamilySet,
+    /// The peer's number in the routing table: what its routes there are
+    /// marked with.
+    pub rib_peer: PeerIndex,
+    outbound: mpsc::Sender<Message>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+#[error("session with {0} has closed")]
+pub struct SessionClosed(pub PeerId);
+
+impl SessionHandle {
+    pub(crate) fn new(
+        peer: PeerId,
+        name: Option<String>,
+        direction: Direction,
+        local_addr: SocketAddr,
+        negotiated: &Negotiated,
+        rib_peer: PeerIndex,
+        outbound: mpsc::Sender<Message>,
+    ) -> Self {
+        Self {
+            peer,
+            name,
+            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(),
+            add_path_send: negotiated.add_path_send.clone(),
+            rib_peer,
+            outbound,
+        }
+    }
+
+    /// Whether `other` is a handle to this same session, and not to another
+    /// session with the same peer.
+    pub fn is_same_session(&self, other: &SessionHandle) -> bool {
+        self.outbound.same_channel(&other.outbound)
+    }
+
+    /// Queues a message for transmission on this session. Waits for queue
+    /// space if the session's writer is behind.
+    pub async fn send(&self, msg: Message) -> Result<(), SessionClosed> {
+        self.outbound
+            .send(msg)
+            .await
+            .map_err(|_| SessionClosed(self.peer))
+    }
+}
+
+#[derive(Debug, Clone)]
+pub enum SessionEvent {
+    Up(SessionHandle),
+    Down(PeerId),
+    /// The peer sent ROUTE-REFRESH (RFC 2918): it wants everything we
+    /// advertise to it sent again, typically because its import policy
+    /// changed.
+    RefreshRequested(SessionHandle),
+}
+
+#[derive(Clone)]
+pub struct SessionRegistry {
+    inner: Arc<RwLock<HashMap<PeerId, SessionHandle>>>,
+    configured: Arc<Vec<ConfiguredPeer>>,
+    dial_states: Arc<RwLock<HashMap<String, DialState>>>,
+    shutting_down: Arc<AtomicBool>,
+    events: broadcast::Sender<SessionEvent>,
+}
+
+impl Default for SessionRegistry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl SessionRegistry {
+    pub fn new() -> Self {
+        Self::with_configured(vec![])
+    }
+
+    pub fn with_configured(configured: Vec<ConfiguredPeer>) -> Self {
+        let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
+        let dial_states = configured
+            .iter()
+            .map(|p| (p.name.clone(), DialState::Idle))
+            .collect();
+        Self {
+            inner: Arc::new(RwLock::new(HashMap::new())),
+            configured: Arc::new(configured),
+            dial_states: Arc::new(RwLock::new(dial_states)),
+            shutting_down: Arc::new(AtomicBool::new(false)),
+            events,
+        }
+    }
+
+    /// Peers from the configuration file, each with the dialer's current
+    /// state for it.
+    pub fn configured_peers(&self) -> Vec<(ConfiguredPeer, DialState)> {
+        let states = self.dial_states.reading();
+        self.configured
+            .iter()
+            .map(|p| {
+                (
+                    p.clone(),
+                    states.get(&p.name).cloned().unwrap_or(DialState::Idle),
+                )
+            })
+            .collect()
+    }
+
+    pub(crate) fn set_dial_state(&self, name: &str, state: DialState) {
+        self.dial_states.writing().insert(name.to_string(), state);
+    }
+
+    /// An established session with this remote IP and ASN, regardless of
+    /// which side opened the TCP connection.
+    pub fn find(&self, ip: IpAddr, asn: u32) -> Option<SessionHandle> {
+        self.inner
+            .reading()
+            .values()
+            .find(|h| h.peer.addr.ip() == ip && h.peer.asn == asn)
+            .cloned()
+    }
+
+    /// Adds a session and emits [`SessionEvent::Up`]. The returned guard
+    /// removes the session again when dropped.
+    pub(crate) fn register(&self, handle: SessionHandle) -> SessionGuard {
+        let peer = handle.peer;
+        self.inner.writing().insert(peer, handle.clone());
+        SESSIONS_ESTABLISHED.inc();
+        let _ = self.events.send(SessionEvent::Up(handle));
+        SessionGuard {
+            registry: self.clone(),
+            peer,
+        }
+    }
+
+    /// 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(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));
+        }
+    }
+
+    /// True once [`shutdown`](Self::shutdown) has begun; dialers stop
+    /// reconnecting.
+    pub fn is_shutting_down(&self) -> bool {
+        self.shutting_down.load(Ordering::Relaxed)
+    }
+
+    /// Closes every session politely: each peer is sent Cease /
+    /// Administrative Shutdown (RFC 4486) so it can drop the session at once
+    /// instead of discovering a dead TCP connection. Waits up to `grace` for
+    /// the sessions to finish; returns how many were still open.
+    pub async fn shutdown(&self, grace: Duration) -> usize {
+        self.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
+                .outbound
+                .try_send(Message::Notification(NotificationMessage {
+                    error: BgpError::Cease(CeaseSubcode::AdminShutdown),
+                    data: vec![],
+                }));
+        }
+        let deadline = tokio::time::Instant::now() + grace;
+        while !self.list().is_empty() && tokio::time::Instant::now() < deadline {
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        self.list().len()
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<SessionEvent> {
+        self.events.subscribe()
+    }
+
+    fn remove(&self, peer: &PeerId) {
+        let removed = self.inner.writing().remove(peer);
+        if removed.is_some() {
+            SESSIONS_ESTABLISHED.dec();
+            let _ = self.events.send(SessionEvent::Down(*peer));
+        }
+    }
+}
+
+/// Removes the session from its registry on drop.
+pub struct SessionGuard {
+    registry: SessionRegistry,
+    peer: PeerId,
+}
+
+impl Drop for SessionGuard {
+    fn drop(&mut self) {
+        self.registry.remove(&self.peer);
+    }
+}