From a772322d9ee3e20f609989ea034deede9a5b0f81 Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Thu, 1 Oct 2026 23:24:00 +0200 Subject: [PATCH] Unwind panics safely: say goodbye to the peer, keep the other sessions A panic is a bug, and it should cost as little as it can. The server already closed every session with a Cease and exited when a RIB or a background task was lost. This fills in the rest: - A panic inside a session is caught. The peer is sent Cease / Administrative Reset, the session's routes are removed in the usual order, and the dialer reconnects. Before, the peer was left with a TCP connection closing under it and no reason. - The tables every session shares (the session registry and the collision table) no longer refuse to be locked after a panic. Lock poisoning would have made each remaining session panic in turn the next time it touched one, so one bug in one session took all of them down. - The server's watcher reports a task that panicked with the panic's message, and is now tested for both a panicking task and a stopped RIB. - The daemon sends panics to its log, with their location, instead of leaving them to the default report on stderr. Co-Authored-By: Claude Fable 5.1 --- bin/bgp/src/main.rs | 23 ++++++ crates/server/src/collision.rs | 7 +- crates/server/src/connection.rs | 124 +++++++++++++++++++++++++------- crates/server/src/error.rs | 7 ++ crates/server/src/lib.rs | 2 + crates/server/src/panic.rs | 60 ++++++++++++++++ crates/server/src/server.rs | 43 +++++++++++ crates/server/src/session.rs | 35 +++------ crates/server/src/sync.rs | 66 +++++++++++++++++ 9 files changed, 313 insertions(+), 54 deletions(-) create mode 100644 crates/server/src/panic.rs create mode 100644 crates/server/src/sync.rs diff --git a/bin/bgp/src/main.rs b/bin/bgp/src/main.rs index 572969d..5786a44 100644 --- a/bin/bgp/src/main.rs +++ b/bin/bgp/src/main.rs @@ -23,6 +23,7 @@ struct ServerArgs { async fn main() -> Result<()> { let args = ServerArgs::parse(); args.tracing.install()?; + log_panics(); let config_string = std::fs::read_to_string(&args.config).wrap_err("Failed to open config file")?; @@ -70,6 +71,28 @@ async fn main() -> Result<()> { } } +/// Sends panics to the log, with where they happened. A panic is a bug, and +/// the default report goes to stderr in a format of its own, apart from +/// everything else the daemon says. The task that panicked still unwinds; +/// the server deals with what it leaves behind. +fn log_panics() { + std::panic::set_hook(Box::new(|info| { + let location = info + .location() + .map(|l| format!("{}:{}", l.file(), l.line())) + .unwrap_or_else(|| "an unknown location".into()); + let message = info + .payload() + .downcast_ref::<&str>() + .map(|s| s.to_string()) + .or_else(|| info.payload().downcast_ref::().cloned()) + .unwrap_or_else(|| "no message".into()); + // Captured only if RUST_BACKTRACE is set. + let backtrace = std::backtrace::Backtrace::capture(); + error!(%location, %message, %backtrace, "Panic: this is a bug"); + })); +} + /// Resolves with the name of the first termination signal received. async fn shutdown_signal() -> &'static str { #[cfg(unix)] diff --git a/crates/server/src/collision.rs b/crates/server/src/collision.rs index 0226cee..4dd50f7 100644 --- a/crates/server/src/collision.rs +++ b/crates/server/src/collision.rs @@ -26,6 +26,7 @@ use std::sync::{Arc, Mutex}; use tokio::sync::oneshot; use crate::metrics::COLLISIONS; +use crate::sync::IgnorePoison; /// Which side opened the TCP connection. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -102,7 +103,7 @@ impl CollisionTable { remote_id: Ipv4Addr, direction: Direction, ) -> Option { - let mut entries = self.entries.lock().expect("collision table poisoned"); + let mut entries = self.entries.locked(); if let Some(existing) = entries.get_mut(&remote_id) { if new_loses( local_id, @@ -151,7 +152,7 @@ pub struct ConnectionGuard { impl ConnectionGuard { pub fn set_established(&self) { - let mut entries = self.table.entries.lock().expect("collision table poisoned"); + let mut entries = self.table.entries.locked(); if let Some(e) = entries.get_mut(&self.remote_id).filter(|e| e.id == self.id) { e.established = true; } @@ -162,7 +163,7 @@ 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"); + let mut entries = self.table.entries.locked(); if entries .get(&self.remote_id) .is_some_and(|e| e.id == self.id) diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index 5953231..301b8d3 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -20,7 +20,7 @@ use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate, RibStopped}; use bytes::BytesMut; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; -use tokio::net::tcp::OwnedWriteHalf; +use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; use tokio::sync::mpsc; use tokio::time::{Instant, Interval, Sleep, interval_at, sleep}; use tracing::{error, info, warn}; @@ -32,6 +32,7 @@ use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated}; use crate::metrics::{ MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, UPDATE_ERRORS, message_type_name, }; +use crate::panic::contained; use crate::session::{SessionGuard, SessionHandle, SessionRegistry}; /// Number of outbound messages that may be queued for a session before @@ -76,30 +77,12 @@ pub async fn run_session( 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, - families: config.families.clone(), - }); - 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 (mut session, first_actions) = Session::open(stream, peer_addr, config)?; - let result = session.run(first_actions).await; + let result = match contained(session.run(first_actions)).await { + Ok(result) => result, + Err(panic) => Err(session.close_after_panic(panic).await), + }; // Whatever ended the session, routes learned over it are gone // (RFC 4271 §8.2.2). Do this before the registration drops so @@ -120,6 +103,40 @@ pub async fn run_session( result } +impl Session { + /// A session on a connection that has just been opened, and the first + /// things to do on it. + fn open( + stream: TcpStream, + peer_addr: SocketAddr, + config: SessionConfig, + ) -> io::Result<(Self, Vec)> { + 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, + families: config.families.clone(), + }); + let 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, + }; + Ok((session, first_actions)) + } +} + /// What exists only once a session is Established. struct Established { peer_id: PeerId, @@ -203,6 +220,25 @@ struct Session { } impl Session { + /// Says goodbye to the peer after the session's own code panicked. + /// + /// The state machine may be half way through something, so it is not + /// asked what to do. The socket is still good for a NOTIFICATION, which + /// lets the peer drop our routes now and reconnect, instead of waiting + /// out its hold timer on a connection nobody is reading. + async fn close_after_panic(&mut self, panic: String) -> SessionError { + let error = SessionError::Panicked(panic); + error!(peer_addr = %self.peer_addr, err = %error, "Closing session after a panic"); + if let Some(notification) = error.notification() { + let notification = Message::Notification(notification); + // Contained as well: whatever panicked may do so again. + if !matches!(contained(self.send(¬ification)).await, Ok(Ok(()))) { + warn!(peer_addr = %self.peer_addr, "Could not send NOTIFICATION after a panic"); + } + } + error + } + async fn run(&mut self, first_actions: Vec) -> Result<(), SessionError> { let mut actions = first_actions; loop { @@ -596,6 +632,46 @@ mod tests { )); } + /// A bug in the session must not leave the peer talking to a connection + /// nobody reads: it is told the session is being reset. + #[tokio::test] + async fn a_session_that_panicked_still_says_goodbye() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let mut peer = TcpStream::connect(listener.local_addr().unwrap()) + .await + .unwrap(); + let (stream, peer_addr) = listener.accept().await.unwrap(); + let config = SessionConfig { + router_id: Ipv4Addr::new(10, 0, 0, 1), + local_asn: 65000, + hold_time_secs: 90, + adj_rib_in: AdjRibIn::spawn(), + registry: SessionRegistry::new(), + collisions: CollisionTable::new(false), + direction: Direction::Inbound, + peer_name: None, + expected_asn: None, + families: FamilySet::supported(), + }; + let (mut session, _) = Session::open(stream, peer_addr, config).unwrap(); + + let error = session + .close_after_panic("index out of bounds".into()) + .await; + assert_eq!( + error.to_string(), + "the session panicked: index out of bounds" + ); + drop(session); + + // Cease / Administrative Reset, then the end of the stream. + let mut received = Vec::new(); + peer.read_to_end(&mut received).await.unwrap(); + let mut expected = vec![0xff; 16]; + expected.extend_from_slice(&[0, 21, 3, 6, 4]); + assert_eq!(received, expected); + } + /// A session task that is torn down without running its cleanup must /// not leave the peer's routes behind. #[tokio::test] diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index 17b9924..8456930 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -46,6 +46,10 @@ pub enum SessionError { /// its way down; the peer is told so. #[error(transparent)] RibStopped(#[from] RibStopped), + /// A bug: the session's own code panicked. The peer is told the session + /// is being reset, and the session is cleaned up like any other. + #[error("the session panicked: {0}")] + Panicked(String), #[error("no address family in common: we offer {ours}, the peer offers {theirs}")] NoCommonAddressFamily { ours: FamilySet, theirs: FamilySet }, } @@ -74,6 +78,9 @@ impl SessionError { Self::WrongPeerAs { .. } => BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), Self::Collision => BgpError::Cease(CeaseSubcode::ConnCollisionResolution), Self::RibStopped(_) => BgpError::Cease(CeaseSubcode::AdminShutdown), + // The session will be set up again; nothing is wrong at the + // peer's end. + Self::Panicked(_) => BgpError::Cease(CeaseSubcode::AdminReset), // RFC 5492 §5: the Data lists the capabilities we need and the // peer did not offer. Self::NoCommonAddressFamily { ours, .. } => { diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index d4f06ff..350f958 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -7,8 +7,10 @@ pub mod error; pub mod family; pub mod fsm; pub mod metrics; +mod panic; pub mod server; pub mod session; +mod sync; pub mod tracing; pub mod proto { diff --git a/crates/server/src/panic.rs b/crates/server/src/panic.rs new file mode 100644 index 0000000..0adf54f --- /dev/null +++ b/crates/server/src/panic.rs @@ -0,0 +1,60 @@ +//! Containing panics. +//! +//! A panic is a bug. It unwinds the task it happens in, and what that task +//! was responsible for still has to be finished properly: a session tells +//! its peer it is closing and gives up its routes; the server, if it lost +//! something it cannot do without, closes every session and exits. + +use std::any::Any; +use std::panic::AssertUnwindSafe; + +use futures::FutureExt; + +/// Runs `future` to completion. If it panics, the panic is caught and its +/// message returned, so that the caller can clean up after it. +/// +/// The caller must not go on using what the future was working on as if +/// nothing had happened: a panic can leave it half changed. Closing it down +/// is what this is for. +pub(crate) async fn contained(future: F) -> Result { + AssertUnwindSafe(future) + .catch_unwind() + .await + .map_err(|payload| message(payload.as_ref())) +} + +/// The text of a panic, from what `catch_unwind` or a `JoinError` carries. +pub(crate) fn message(payload: &(dyn Any + Send)) -> String { + if let Some(text) = payload.downcast_ref::<&'static str>() { + (*text).to_string() + } else if let Some(text) = payload.downcast_ref::() { + text.clone() + } else { + "a panic that carried no message".to_string() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn a_panic_is_caught_with_its_message() { + assert_eq!(contained(async { 7 }).await, Ok(7)); + + let literal: Result<(), _> = contained(async { panic!("the table is gone") }).await; + assert_eq!(literal, Err("the table is gone".to_string())); + + let formatted: Result<(), _> = + contained(async { panic!("index {} out of range", 9) }).await; + assert_eq!(formatted, Err("index 9 out of range".to_string())); + + // After a point where the future had been suspended, too. + let later: Result<(), _> = contained(async { + tokio::task::yield_now().await; + panic!("later") + }) + .await; + assert_eq!(later, Err("later".to_string())); + } +} diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index 164daf7..62f765c 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -18,6 +18,7 @@ use crate::connection::{SessionConfig, handle_connection}; use crate::dialer::spawn_dialers; use crate::error::ServerError; use crate::family::FamilySet; +use crate::panic; use crate::session::{ConfiguredPeer, SessionRegistry}; pub struct BgpServerBuilder { @@ -218,6 +219,9 @@ async fn failed( task, reason: match result { Ok(()) => "it returned".into(), + Err(e) if e.is_panic() => { + format!("it panicked: {}", panic::message(e.into_panic().as_ref())) + } Err(e) => e.to_string(), }, }, @@ -231,3 +235,42 @@ async fn accept_one( let result = listener.accept().await; (addr, result, listener) } + +#[cfg(test)] +mod tests { + use super::*; + + /// The advertiser and the dialers run for as long as the server does. + /// If one of them panics, nothing else would ever find out. + #[tokio::test] + async fn a_task_that_panics_is_a_failure_of_the_server() { + let adj_rib_in = AdjRibIn::spawn(); + let local_rib = LocalRib::spawn(); + let mut tasks: FuturesUnordered = FuturesUnordered::new(); + tasks.push(named("dialer", tokio::spawn(std::future::pending()))); + tasks.push(named( + "advertiser", + tokio::spawn(async { panic!("no route to announce") }), + )); + + let failure = failed(&adj_rib_in, &local_rib, &mut tasks).await; + assert_eq!( + failure.to_string(), + "the advertiser task stopped unexpectedly: it panicked: no route to announce" + ); + } + + #[tokio::test] + async fn a_stopped_rib_is_a_failure_of_the_server() { + let adj_rib_in = AdjRibIn::spawn(); + let local_rib = LocalRib::spawn(); + let mut tasks: FuturesUnordered = FuturesUnordered::new(); + tasks.push(named("advertiser", tokio::spawn(std::future::pending()))); + + local_rib.stop_for_testing(); + let failure = failed(&adj_rib_in, &local_rib, &mut tasks).await; + assert!( + matches!(failure, ServerError::RibStopped(RibStopped { rib }) if rib == "local RIB") + ); + } +} diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index ed707b6..a5f04ec 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -21,6 +21,7 @@ use tokio::sync::{broadcast, mpsc}; use crate::collision::Direction; use crate::family::FamilySet; use crate::metrics::SESSIONS_ESTABLISHED; +use crate::sync::IgnorePoisonRw; const EVENT_CHANNEL_CAPACITY: usize = 256; @@ -150,7 +151,7 @@ impl SessionRegistry { /// 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.read().expect("dial states poisoned"); + let states = self.dial_states.reading(); self.configured .iter() .map(|p| { @@ -168,18 +169,14 @@ impl SessionRegistry { } pub fn set_dial_state(&self, name: &str, state: DialState) { - self.dial_states - .write() - .expect("dial states poisoned") - .insert(name.to_string(), state); + 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 { self.inner - .read() - .expect("session registry poisoned") + .reading() .values() .find(|h| h.peer.addr.ip() == ip && h.peer.asn == asn) .cloned() @@ -189,10 +186,7 @@ impl SessionRegistry { /// removes the session again when dropped. pub fn register(&self, handle: SessionHandle) -> SessionGuard { let peer = handle.peer; - self.inner - .write() - .expect("session registry poisoned") - .insert(peer, handle.clone()); + self.inner.writing().insert(peer, handle.clone()); SESSIONS_ESTABLISHED.inc(); let _ = self.events.send(SessionEvent::Up(handle)); SessionGuard { @@ -202,21 +196,12 @@ impl SessionRegistry { } pub fn get(&self, peer: &PeerId) -> Option { - self.inner - .read() - .expect("session registry poisoned") - .get(peer) - .cloned() + self.inner.reading().get(peer).cloned() } /// All currently established sessions, in no particular order. pub fn list(&self) -> Vec { - self.inner - .read() - .expect("session registry poisoned") - .values() - .cloned() - .collect() + self.inner.reading().values().cloned().collect() } /// Called by a session when its peer asks for a route refresh. @@ -258,11 +243,7 @@ impl SessionRegistry { } fn remove(&self, peer: &PeerId) { - let removed = self - .inner - .write() - .expect("session registry poisoned") - .remove(peer); + let removed = self.inner.writing().remove(peer); if removed.is_some() { SESSIONS_ESTABLISHED.dec(); let _ = self.events.send(SessionEvent::Down(*peer)); diff --git a/crates/server/src/sync.rs b/crates/server/src/sync.rs new file mode 100644 index 0000000..fa55249 --- /dev/null +++ b/crates/server/src/sync.rs @@ -0,0 +1,66 @@ +//! Locks that survive a panic. +//! +//! A standard lock is poisoned when a thread panics while holding it, and +//! every later `lock().unwrap()` then panics too. For the tables shared by +//! all sessions that would turn a bug in one session into the end of every +//! session, one panic at a time. +//! +//! The tables here are maps that are changed by single inserts and removes, +//! so a panic part-way through an update cannot leave one in a state that +//! is unsafe to read. The poison is therefore ignored: a session that +//! panics is cleaned up by its guards like one that ended any other way. + +use std::sync::{Mutex, MutexGuard, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard}; + +pub(crate) trait IgnorePoison { + /// Locks, whether or not an earlier holder panicked. + fn locked(&self) -> MutexGuard<'_, T>; +} + +impl IgnorePoison for Mutex { + fn locked(&self) -> MutexGuard<'_, T> { + self.lock().unwrap_or_else(PoisonError::into_inner) + } +} + +pub(crate) trait IgnorePoisonRw { + /// Locks for reading, whether or not an earlier holder panicked. + fn reading(&self) -> RwLockReadGuard<'_, T>; + /// Locks for writing, whether or not an earlier holder panicked. + fn writing(&self) -> RwLockWriteGuard<'_, T>; +} + +impl IgnorePoisonRw for RwLock { + fn reading(&self) -> RwLockReadGuard<'_, T> { + self.read().unwrap_or_else(PoisonError::into_inner) + } + fn writing(&self) -> RwLockWriteGuard<'_, T> { + self.write().unwrap_or_else(PoisonError::into_inner) + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use super::*; + + #[test] + fn a_lock_whose_holder_panicked_can_still_be_taken() { + let mutex = Arc::new(Mutex::new(1)); + let rwlock = Arc::new(RwLock::new(1)); + let (m, r) = (mutex.clone(), rwlock.clone()); + let panicked = std::thread::spawn(move || { + let _m = m.lock().unwrap(); + let _r = r.write().unwrap(); + panic!("while holding both"); + }) + .join(); + assert!(panicked.is_err()); + assert!(mutex.is_poisoned() && rwlock.is_poisoned()); + + *mutex.locked() += 1; + *rwlock.writing() += 1; + assert_eq!((*mutex.locked(), *rwlock.reading()), (2, 2)); + } +}