From af76ecb34522b53424a756bce737fea689d50d6f Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Sun, 20 Sep 2026 17:51:04 +0200 Subject: [PATCH] fix: leave sessions politely on shutdown; answer ROUTE-REFRESH Against a production BIRD router, restarting ferrix was followed by several refused connection attempts before the session came back. - The daemon had no signal handling: Ctrl-C killed the process and the peer only ever saw a dead TCP connection. On SIGINT or SIGTERM it now sends Cease / Administrative Shutdown (RFC 4486) to every established peer, waits up to two seconds for the sessions to close, and stops the dialers from reconnecting meanwhile. A NOTIFICATION queued for a session ends that session once sent. - ROUTE-REFRESH was parsed and ignored. A peer sends it when its import policy changes (e.g. after `birdc configure`), expecting our routes again so the new policy can be applied. The session now reports the request and the advertiser re-sends every local route to that peer. The request also appears in the event stream and the status page log. Tests: a refresh request in real wire format re-sends all routes and keeps the session up; shutdown delivers the Cease to every peer and closes the sessions. Verified with the real binary: SIGINT, peer logs the Cease. Co-Authored-By: Claude Fable 5.1 --- bin/bgp/src/main.rs | 37 ++++++++++- crates/integration-tests/tests/announce.rs | 72 ++++++++++++++++++++++ crates/server/src/advertise.rs | 7 +++ crates/server/src/dialer.rs | 3 + crates/server/src/fsm.rs | 13 +++- crates/server/src/session.rs | 46 +++++++++++++- crates/web/src/events.rs | 3 + crates/web/static/statusz.html | 1 + docs/RFCs.md | 3 +- 9 files changed, 179 insertions(+), 6 deletions(-) diff --git a/bin/bgp/src/main.rs b/bin/bgp/src/main.rs index b4befc2..8b2a7bc 100644 --- a/bin/bgp/src/main.rs +++ b/bin/bgp/src/main.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::time::Duration; use bgp_server::config::Config; use bgp_server::proto::config::BgpConfig; @@ -7,7 +8,7 @@ use bgp_server::tracing::TracingArgs; use bgp_web::{AppState, RouterInfo}; use clap::Parser; use eyre::{Context, Result}; -use tracing::error; +use tracing::{error, info, warn}; #[derive(Parser, Debug)] struct ServerArgs { @@ -52,5 +53,37 @@ async fn main() -> Result<()> { }); } - server.run().await + // Leave politely: tell every peer we are going (Cease / Administrative + // Shutdown) rather than letting them find a dead TCP connection, which + // makes them slower to take us back. + let registry = server.registry.clone(); + tokio::select! { + result = server.run() => 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"); + } + Ok(()) + } + } +} + +/// Resolves with the name of the first termination signal received. +async fn shutdown_signal() -> &'static str { + #[cfg(unix)] + { + use tokio::signal::unix::{SignalKind, signal}; + let mut term = signal(SignalKind::terminate()).expect("install SIGTERM handler"); + tokio::select! { + _ = tokio::signal::ctrl_c() => "SIGINT", + _ = term.recv() => "SIGTERM", + } + } + #[cfg(not(unix))] + { + let _ = tokio::signal::ctrl_c().await; + "ctrl-c" + } } diff --git a/crates/integration-tests/tests/announce.rs b/crates/integration-tests/tests/announce.rs index c778a7d..34d07ff 100644 --- a/crates/integration-tests/tests/announce.rs +++ b/crates/integration-tests/tests/announce.rs @@ -279,6 +279,78 @@ async fn test_received_routes_removed_when_session_closes() -> Result<()> { Ok(()) } +// --------------------------------------------------------------------------- +// Route refresh and shutdown +// --------------------------------------------------------------------------- + +/// ROUTE-REFRESH for IPv4 unicast exactly as a router sends it (RFC 2918): +/// 19-byte header plus AFI, reserved, SAFI. +#[rustfmt::skip] +const ROUTE_REFRESH_IPV4: &[u8] = &[ + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x17, 0x05, + 0x00, 0x01, 0x00, 0x01, +]; + +/// A peer whose import policy changed asks for our routes again; it must get +/// every local route re-sent, and the session must stay up. +#[tokio::test] +async fn test_route_refresh_resends_local_routes() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut peer = established_peer(&server).await?; + wait_for_sessions(&server, 1).await?; + + server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?; + server.local_rib.announce(route("10.100.2.0/24", LocalRouteAttrs::default())).await?; + let _ = next_update(&mut peer).await?; + let _ = next_update(&mut peer).await?; + + peer.send_raw(ROUTE_REFRESH_IPV4).await?; + + let mut resent = vec![next_update(&mut peer).await?, next_update(&mut peer).await?] + .into_iter() + .flat_map(|u| u.announced_routes) + .map(|p| p.to_string()) + .collect::>(); + resent.sort(); + assert_eq!(resent, vec!["10.100.1.0/24", "10.100.2.0/24"]); + assert_eq!(server.registry.list().len(), 1, "session must survive a route refresh"); + Ok(()) +} + +/// Shutting down tells each peer why (Cease / Administrative Shutdown) and +/// closes the sessions, instead of leaving dead TCP connections behind. +#[tokio::test] +async fn test_shutdown_sends_cease_to_every_peer() -> Result<()> { + use bgp_packet::errors::{BgpError, CeaseSubcode}; + use bgp_packet::message::Message; + + let server = TestServer::start(LOCAL_ASN).await?; + let mut a = established_peer_with_id(&server, Ipv4Addr::new(10, 0, 0, 1)).await?; + let mut b = established_peer_with_id(&server, Ipv4Addr::new(10, 0, 0, 2)).await?; + wait_for_sessions(&server, 2).await?; + + let left = server.registry.shutdown(Duration::from_secs(2)).await; + assert_eq!(left, 0, "sessions still open after shutdown"); + assert!(server.registry.is_shutting_down()); + + for peer in [&mut a, &mut b] { + let notif = tokio::time::timeout(TIMEOUT, async { + loop { + match peer.recv().await? { + Message::Notification(n) => return Ok::<_, eyre::Error>(n), + Message::KeepAlive => {} + other => eyre::bail!("unexpected {other:?}"), + } + } + }) + .await??; + assert!(matches!(notif.error, BgpError::Cease(CeaseSubcode::AdminShutdown)), "{notif:?}"); + assert!(peer.is_closed().await); + } + Ok(()) +} + // --------------------------------------------------------------------------- // Policy // --------------------------------------------------------------------------- diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 3be879a..83f327c 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -62,6 +62,13 @@ async fn run(local_rib: LocalRib, registry: SessionRegistry, local_asn: u32) { send_route(&session, route, local_asn).await; } } + Ok(SessionEvent::RefreshRequested(session)) => { + let routes = local_rib.snapshot().await; + 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; + } + } Ok(SessionEvent::Down(_)) => {} Err(RecvError::Lagged(n)) => { warn!(missed = n, "Advertiser lagged on session events, resyncing all sessions"); diff --git a/crates/server/src/dialer.rs b/crates/server/src/dialer.rs index b15fe14..5a6910b 100644 --- a/crates/server/src/dialer.rs +++ b/crates/server/src/dialer.rs @@ -49,6 +49,9 @@ async fn dial_loop(peer: Arc, config: SessionConfig, connect_retry: let mut backoff = connect_retry; loop { + if registry.is_shutting_down() { + return; + } if registry.find(peer.endpoint.ip(), peer.remote_asn).is_some() { // Peer dialled us; nothing to do until that session goes away. registry.set_dial_state(&peer.name, DialState::Idle); diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index b834e6f..9fcb6a1 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -204,7 +204,7 @@ pub async fn run_session( negotiated_hold_time, peer_id, &config.adj_rib_in, - SessionInputs { outbound: &mut outbound_rx, killed: &mut collision_guard.killed }, + SessionInputs { outbound: &mut outbound_rx, killed: &mut collision_guard.killed, registry: &config.registry }, ) .await; @@ -222,6 +222,8 @@ struct SessionInputs<'a> { outbound: &'a mut mpsc::Receiver, /// Fires when a later connection won collision resolution. killed: &'a mut oneshot::Receiver<()>, + /// Where to report a peer's route-refresh request. + registry: &'a SessionRegistry, } async fn established_loop( @@ -233,7 +235,7 @@ async fn established_loop( adj_rib_in: &AdjRibIn, inputs: SessionInputs<'_>, ) -> Result<()> { - let SessionInputs { outbound, killed } = inputs; + let SessionInputs { outbound, killed, registry } = 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. @@ -277,6 +279,8 @@ async fn established_loop( } Message::RouteRefresh => { hold_timer.as_mut().reset(Instant::now() + hold_duration); + info!(peer = %peer_id, "Received ROUTE-REFRESH, re-sending our routes"); + registry.request_refresh(&peer_id); } } } @@ -290,6 +294,11 @@ async fn established_loop( ); } send_message(writer, &msg, ctx).await?; + // A NOTIFICATION from our side ends the session (RFC 4271 §6). + if let Message::Notification(notif) = &msg { + info!(peer = %peer_id, error = %notif.error, "Sent NOTIFICATION, closing session"); + return Ok(()); + } } _ = &mut *killed => { return close_for_collision(writer, ctx, peer_id.addr).await; diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index 5162362..b20b840 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -7,8 +7,12 @@ use std::collections::HashMap; use std::net::{IpAddr, SocketAddr}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, RwLock}; -use std::time::SystemTime; +use std::time::{Duration, SystemTime}; + +use bgp_packet::errors::{BgpError, CeaseSubcode}; +use bgp_packet::message::NotificationMessage; use bgp_packet::message::Message; use bgp_rib::PeerId; @@ -82,6 +86,10 @@ impl SessionHandle { 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)] @@ -89,6 +97,7 @@ pub struct SessionRegistry { inner: Arc>>, configured: Arc>, dial_states: Arc>>, + shutting_down: Arc, events: broadcast::Sender, } @@ -110,6 +119,7 @@ impl SessionRegistry { 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, } } @@ -163,6 +173,40 @@ impl SessionRegistry { self.inner.read().expect("session registry poisoned").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) { + 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() { + let _ = session + .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 { + tokio::time::sleep(Duration::from_millis(20)).await; + } + self.list().len() + } + pub fn subscribe(&self) -> broadcast::Receiver { self.events.subscribe() } diff --git a/crates/web/src/events.rs b/crates/web/src/events.rs index e9ec8ed..6c20676 100644 --- a/crates/web/src/events.rs +++ b/crates/web/src/events.rs @@ -20,6 +20,8 @@ use crate::api::{LocalRouteView, PeerView, ReceivedRouteView, summarize}; pub enum WireEvent { SessionUp { peer: PeerView }, SessionDown { peer_addr: String, peer_asn: u32 }, + /// The peer asked us to send all our routes again (ROUTE-REFRESH). + RouteRefreshRequested { peer: PeerView }, LocalAnnounced { route: LocalRouteView }, LocalWithdrawn { prefix: String }, ReceivedAnnounced { route: ReceivedRouteView }, @@ -34,6 +36,7 @@ impl WireEvent { match ev { Ok(SessionEvent::Up(h)) => Self::SessionUp { peer: PeerView::from(&h) }, Ok(SessionEvent::Down(p)) => Self::SessionDown { peer_addr: p.addr.to_string(), peer_asn: p.asn }, + Ok(SessionEvent::RefreshRequested(h)) => Self::RouteRefreshRequested { peer: PeerView::from(&h) }, Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged { stream: "sessions", missed: n }, } } diff --git a/crates/web/static/statusz.html b/crates/web/static/statusz.html index 9d3f5c8..36511aa 100644 --- a/crates/web/static/statusz.html +++ b/crates/web/static/statusz.html @@ -407,6 +407,7 @@ View: variables, status json, < function describe(ev) { switch (ev.type) { case 'session_up': return `session UP with ${esc(ev.peer.name || ev.peer.peer_addr)} AS${ev.peer.peer_asn} (opened by ${ev.peer.direction === 'outbound' ? 'us' : 'peer'})`; + case 'route_refresh_requested': return `${esc(ev.peer.name || ev.peer.peer_addr)} AS${ev.peer.peer_asn} asked for a route refresh; re-sent all local routes`; case 'session_down': return `session DOWN with ${esc(ev.peer_addr)} AS${ev.peer_asn}`; case 'local_announced': { const r = ev.route; return `announced ${esc(r.prefix)} by ${esc(r.owner)}` + (r.communities.length ? ` [${names(r.communities)}]` : '') + (r.large_communities.length ? ` [${names(r.large_communities)}]` : '') + diff --git a/docs/RFCs.md b/docs/RFCs.md index 16a97c9..832c0e0 100644 --- a/docs/RFCs.md +++ b/docs/RFCs.md @@ -27,6 +27,7 @@ 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.7 / RFC 4486 | Cease / Administrative Shutdown sent to every peer on SIGINT or SIGTERM | ✅ | | §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) | ✅ | @@ -66,7 +67,7 @@ https://datatracker.ietf.org/doc/html/rfc2918 | `RouteRefresh` capability advertisement | ✅ | | `ROUTE-REFRESH` message type (5) framing | ✅ | | AFI/SAFI payload parsing | ❌ | -| Acting on received ROUTE-REFRESH (re-advertising routes) | ❌ | +| Acting on received ROUTE-REFRESH (re-advertising all local routes, regardless of the AFI/SAFI asked for) | ✅ | ### RFC 7313 — Enhanced Route Refresh Capability for BGP-4 https://datatracker.ietf.org/doc/html/rfc7313