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 <noreply@anthropic.com>

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::<String>().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)]

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<ConnectionGuard> {
-        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)

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<OwnedReadHalf> {
+    /// 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<Action>)> {
+        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<R> {
 }
 
 impl<R: AsyncRead + Unpin> Session<R> {
+    /// 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(&notification)).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<Action>) -> 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]

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, .. } => {

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 {

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<F: Future>(future: F) -> Result<F::Output, String> {
+    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::<String>() {
+        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()));
+    }
+}

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<NamedTask> = 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<NamedTask> = 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")
+        );
+    }
+}

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<SessionHandle> {
         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<SessionHandle> {
-        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<SessionHandle> {
-        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));

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<T> {
+    /// Locks, whether or not an earlier holder panicked.
+    fn locked(&self) -> MutexGuard<'_, T>;
+}
+
+impl<T> IgnorePoison<T> for Mutex<T> {
+    fn locked(&self) -> MutexGuard<'_, T> {
+        self.lock().unwrap_or_else(PoisonError::into_inner)
+    }
+}
+
+pub(crate) trait IgnorePoisonRw<T> {
+    /// 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<T> IgnorePoisonRw<T> for RwLock<T> {
+    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));
+    }
+}