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 <noreply@anthropic.com>
This commit is contained in:
Rayhaan JaufeerallyandClaude Fable 5.1 committed 2026-09-20 17:51:04 +02:00
1 parent 57099da73d
commit af76ecb345
9 files changed
+179 -6

No files matched your search

+35 -2
View File
@@ -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"
}
}
@@ -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::<Vec<_>>();
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
// ---------------------------------------------------------------------------
+7
View File
@@ -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");
+3
View File
@@ -49,6 +49,9 @@ async fn dial_loop(peer: Arc<PeerConfig>, 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);
+11 -2
View File
@@ -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<Message>,
/// 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;
+45 -1
View File
@@ -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<RwLock<HashMap<PeerId, SessionHandle>>>,
configured: Arc<Vec<ConfiguredPeer>>,
dial_states: Arc<RwLock<HashMap<String, DialState>>>,
shutting_down: Arc<AtomicBool>,
events: broadcast::Sender<SessionEvent>,
}
@@ -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<SessionEvent> {
self.events.subscribe()
}
+3
View File
@@ -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 },
}
}
+1
View File
@@ -407,6 +407,7 @@ View: <a href="/metricz">variables</a>, <a href="/api/status">status json</a>, <
function describe(ev) {
switch (ev.type) {
case 'session_up': return `<span class="ok">session UP</span> 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 `<span class="bad">session DOWN</span> 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)}]` : '') +
+2 -1
View File
@@ -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