diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index d0a956a..c4cad15 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -75,6 +75,14 @@ impl TestServer { } impl TestServer { + /// Waits for the server to stop by itself and returns why. A healthy + /// server never does. + pub async fn stopped(&mut self) -> Result<(), ServerError> { + (&mut self.handle) + .await + .expect("the server task does not panic") + } + /// Starts the HTTP API on an ephemeral loopback port and returns its /// base URL (e.g. `http://127.0.0.1:12345`). The API task is aborted /// when the returned guard drops. diff --git a/crates/integration-tests/tests/announce.rs b/crates/integration-tests/tests/announce.rs index e3431ea..3685c00 100644 --- a/crates/integration-tests/tests/announce.rs +++ b/crates/integration-tests/tests/announce.rs @@ -306,7 +306,7 @@ async fn test_received_routes_removed_when_session_closes() -> Result<()> { let ev = tokio::time::timeout(TIMEOUT, rib_events.recv()).await??; assert!(matches!(ev, bgp_rib::RibEvent::Announced { .. }), "{ev:?}"); } - assert_eq!(server.adj_rib_in.snapshot().await.len(), 2); + assert_eq!(server.adj_rib_in.snapshot().await.unwrap().len(), 2); drop(peer); @@ -319,7 +319,7 @@ async fn test_received_routes_removed_when_session_closes() -> Result<()> { } withdrawn.sort(); assert_eq!(withdrawn, vec!["198.51.100.0/24", "203.0.113.0/24"]); - assert!(server.adj_rib_in.snapshot().await.is_empty()); + assert!(server.adj_rib_in.snapshot().await.unwrap().is_empty()); // Withdrawals precede the Down event. let _up = tokio::time::timeout(TIMEOUT, session_events.recv()).await??; @@ -431,7 +431,7 @@ async fn test_configured_routes_are_announced_to_every_new_session() -> Result<( route("10.100.2.0/24", LocalRouteAttrs::default()), ]; let server = TestServer::start_with(LOCAL_ASN, |c| c.originate = configured).await?; - assert_eq!(server.local_rib.snapshot().await.len(), 2); + assert_eq!(server.local_rib.snapshot().await.unwrap().len(), 2); for router_id in [Ipv4Addr::new(10, 0, 0, 1), Ipv4Addr::new(10, 0, 0, 2)] { let mut peer = established_peer_with_id(&server, router_id).await?; diff --git a/crates/integration-tests/tests/basic_connection.rs b/crates/integration-tests/tests/basic_connection.rs index e089ef3..be5ba95 100644 --- a/crates/integration-tests/tests/basic_connection.rs +++ b/crates/integration-tests/tests/basic_connection.rs @@ -233,6 +233,7 @@ async fn test_message_split_across_a_keepalive_tick_is_still_one_message() -> Re .adj_rib_in .snapshot() .await + .unwrap() .iter() .any(|e| e.prefix == prefix) { @@ -412,7 +413,7 @@ async fn test_rib_records_ipv4_announce() -> Result<()> { "Expected Announced for 10.0.0.0/8, got {event:?}", ); - let snapshot = server.adj_rib_in.snapshot().await; + let snapshot = server.adj_rib_in.snapshot().await.unwrap(); assert!( snapshot.iter().any(|e| e.prefix == prefix), "10.0.0.0/8 missing from RIB snapshot", @@ -455,7 +456,7 @@ async fn test_rib_records_ipv4_withdraw() -> Result<()> { "Expected Withdrawn for 10.0.0.0/8, got {event:?}", ); - let snapshot = server.adj_rib_in.snapshot().await; + let snapshot = server.adj_rib_in.snapshot().await.unwrap(); assert!( !snapshot.iter().any(|e| e.prefix == prefix), "10.0.0.0/8 should have been removed from RIB snapshot", @@ -494,7 +495,7 @@ async fn test_rib_records_ipv6_announce() -> Result<()> { "Expected Announced for 2001:db8::/32, got {event:?}", ); - let snapshot = server.adj_rib_in.snapshot().await; + let snapshot = server.adj_rib_in.snapshot().await.unwrap(); assert!( snapshot.iter().any(|e| e.prefix == prefix), "2001:db8::/32 missing from RIB snapshot", @@ -548,7 +549,7 @@ async fn test_rib_records_ipv6_withdraw() -> Result<()> { "Expected Withdrawn for 2001:db8::/32, got {event:?}", ); - let snapshot = server.adj_rib_in.snapshot().await; + let snapshot = server.adj_rib_in.snapshot().await.unwrap(); assert!( !snapshot.iter().any(|e| e.prefix == prefix), "2001:db8::/32 should have been removed from RIB snapshot", diff --git a/crates/integration-tests/tests/collision.rs b/crates/integration-tests/tests/collision.rs index 6022121..aeffc9c 100644 --- a/crates/integration-tests/tests/collision.rs +++ b/crates/integration-tests/tests/collision.rs @@ -240,7 +240,7 @@ async fn test_restarted_peer_replaces_stale_session_when_enabled() -> Result<()> })) .await?; tokio::time::timeout(TIMEOUT, async { - while server.adj_rib_in.snapshot().await.is_empty() { + while server.adj_rib_in.snapshot().await.unwrap().is_empty() { tokio::time::sleep(Duration::from_millis(10)).await; } }) @@ -265,7 +265,7 @@ async fn test_restarted_peer_replaces_stale_session_when_enabled() -> Result<()> }) .await?; assert!( - server.adj_rib_in.snapshot().await.is_empty(), + server.adj_rib_in.snapshot().await.unwrap().is_empty(), "stale routes were not flushed" ); Ok(()) diff --git a/crates/integration-tests/tests/failure.rs b/crates/integration-tests/tests/failure.rs new file mode 100644 index 0000000..0a49ebd --- /dev/null +++ b/crates/integration-tests/tests/failure.rs @@ -0,0 +1,107 @@ +//! What the server does when a part of it that it cannot do without fails. +//! +//! A RIB's actor stops only by panicking. Before, nothing noticed: sessions +//! stayed up, routes sent by peers were dropped on the floor, and the status +//! page showed an empty table. Now the server tells its peers it is going +//! and stops with an error, for whatever supervises it to restart it. + +use std::net::Ipv4Addr; +use std::time::Duration; + +use bgp_integration_tests::test_peer::TestPeer; +use bgp_integration_tests::test_server::TestServer; +use bgp_packet::errors::{BgpError, CeaseSubcode}; +use bgp_packet::message::Message; +use bgp_rib::RibStopped; +use bgp_server::error::ServerError; +use eyre::{Result, bail}; + +const LOCAL_ASN: u32 = 65000; +const PEER_ASN: u32 = 65001; +const TIMEOUT: Duration = Duration::from_secs(5); + +async fn established(server: &TestServer) -> Result { + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; + tokio::time::timeout(TIMEOUT, async { + while server.registry.list().is_empty() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + Ok(peer) +} + +/// Reads until the server's NOTIFICATION, skipping KEEPALIVEs. +async fn farewell(peer: &mut TestPeer) -> Result { + tokio::time::timeout(TIMEOUT, async { + loop { + match peer.recv().await? { + Message::KeepAlive => {} + Message::Notification(n) => return Ok(n.error), + other => bail!("expected NOTIFICATION, got {other:?}"), + } + } + }) + .await? +} + +#[tokio::test] +async fn test_losing_the_adj_rib_in_stops_the_server_and_tells_the_peers() -> Result<()> { + let mut server = TestServer::start(LOCAL_ASN).await?; + let (base, _http) = server.serve_http().await?; + let mut peer = established(&server).await?; + + server.adj_rib_in.stop_for_testing(); + + // Nothing has used the table since; the server notices by itself. + let outcome = tokio::time::timeout(TIMEOUT, server.stopped()).await?; + assert!( + matches!(outcome, Err(ServerError::RibStopped(RibStopped { rib })) if rib == "Adj-RIB-In"), + "{outcome:?}" + ); + assert_eq!( + farewell(&mut peer).await?, + BgpError::Cease(CeaseSubcode::AdminShutdown) + ); + + // The status interface says the table cannot be read, where it used to + // show an empty one. + let client = reqwest::Client::new(); + for path in ["/api/status", "/api/routes/received", "/metricz"] { + let response = client.get(format!("{base}{path}")).send().await?; + assert_eq!(response.status(), 503, "{path}"); + assert_eq!( + response.text().await?, + "the Adj-RIB-In has stopped", + "{path}" + ); + } + // What does not need the table still answers. + let response = client + .get(format!("{base}/api/routes/local")) + .send() + .await?; + assert_eq!(response.status(), 200); + Ok(()) +} + +#[tokio::test] +async fn test_losing_the_local_rib_stops_the_server_and_tells_the_peers() -> Result<()> { + let mut server = TestServer::start(LOCAL_ASN).await?; + let mut peer = established(&server).await?; + + server.local_rib.stop_for_testing(); + + let outcome = tokio::time::timeout(TIMEOUT, server.stopped()).await?; + assert!( + matches!(outcome, Err(ServerError::RibStopped(RibStopped { rib })) if rib == "local RIB"), + "{outcome:?}" + ); + assert_eq!( + farewell(&mut peer).await?, + BgpError::Cease(CeaseSubcode::AdminShutdown) + ); + Ok(()) +} diff --git a/crates/integration-tests/tests/http_api.rs b/crates/integration-tests/tests/http_api.rs index 8f839f9..e602817 100644 --- a/crates/integration-tests/tests/http_api.rs +++ b/crates/integration-tests/tests/http_api.rs @@ -143,7 +143,7 @@ async fn test_api_is_read_only() -> Result<()> { .await?; assert_eq!(delete.status(), 405, "DELETE {path}"); } - assert!(server.local_rib.snapshot().await.is_empty()); + assert!(server.local_rib.snapshot().await.unwrap().is_empty()); Ok(()) } diff --git a/crates/integration-tests/tests/interop.rs b/crates/integration-tests/tests/interop.rs index 6405cec..fd10dd3 100644 --- a/crates/integration-tests/tests/interop.rs +++ b/crates/integration-tests/tests/interop.rs @@ -105,7 +105,7 @@ async fn test_bird_open_and_end_of_rib_are_accepted() -> Result<()> { 1, "session dropped after End-of-RIB" ); - assert!(server.adj_rib_in.snapshot().await.is_empty()); + assert!(server.adj_rib_in.snapshot().await.unwrap().is_empty()); Ok(()) } diff --git a/crates/integration-tests/tests/negotiation.rs b/crates/integration-tests/tests/negotiation.rs index 7752425..5c76703 100644 --- a/crates/integration-tests/tests/negotiation.rs +++ b/crates/integration-tests/tests/negotiation.rs @@ -120,7 +120,7 @@ async fn test_routes_of_a_family_not_negotiated_are_not_accepted() -> Result<()> let routes = tokio::time::timeout(TIMEOUT, async { loop { - let routes = server.adj_rib_in.snapshot().await; + let routes = server.adj_rib_in.snapshot().await.unwrap(); if !routes.is_empty() { return routes; } @@ -304,7 +304,7 @@ async fn test_a_peer_without_capabilities_gets_two_octet_as_paths() -> Result<() .await?; let routes = tokio::time::timeout(TIMEOUT, async { loop { - let routes = server.adj_rib_in.snapshot().await; + let routes = server.adj_rib_in.snapshot().await.unwrap(); if !routes.is_empty() { return routes; } diff --git a/crates/integration-tests/tests/update_errors.rs b/crates/integration-tests/tests/update_errors.rs index baf7c74..f48a93d 100644 --- a/crates/integration-tests/tests/update_errors.rs +++ b/crates/integration-tests/tests/update_errors.rs @@ -75,7 +75,7 @@ async fn wait_for_routes(server: &TestServer, prefixes: &[&str]) -> Result>, }, + Len { + reply: oneshot::Sender, + }, } +const STOPPED: RibStopped = RibStopped { rib: "Adj-RIB-In" }; + /// A cheaply-cloneable handle to the Adj-RIB-In actor. +/// +/// Every operation fails with [`RibStopped`] if the actor is gone; see +/// there for what that means. #[derive(Clone)] pub struct AdjRibIn { tx: mpsc::Sender, events: broadcast::Sender, + actor: AbortHandle, } impl AdjRibIn { @@ -43,37 +53,68 @@ impl AdjRibIn { pub fn spawn() -> Self { let (tx, rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); - tokio::spawn(run_actor(rx, events.clone())); - Self { tx, events } + let actor = tokio::spawn(run_actor(rx, events.clone())).abort_handle(); + Self { tx, events, actor } } /// Applies one UPDATE received from `peer`: its withdrawals first, then /// its announcements, so a prefix named in both ends up announced /// (RFC 4271 §4.3). - pub async fn apply(&self, peer: PeerId, update: ReceivedUpdate) { - let _ = self.tx.send(RibCommand::Apply { peer, update }).await; + /// + /// Returns once the update is queued for the actor, which applies + /// updates in the order they were queued. + pub async fn apply(&self, peer: PeerId, update: ReceivedUpdate) -> Result<(), RibStopped> { + self.tx + .send(RibCommand::Apply { peer, update }) + .await + .map_err(|_| STOPPED) } /// Removes every route learned from `peer`, emitting a /// [`RibEvent::Withdrawn`] for each (RFC 4271 §8.2.2: routes from a /// terminated session are deleted). Returns how many were removed. - pub async fn remove_peer(&self, peer: PeerId) -> usize { - let (reply_tx, reply_rx) = oneshot::channel(); - let _ = self - .tx - .send(RibCommand::RemovePeer { - peer, - reply: reply_tx, - }) - .await; - reply_rx.await.unwrap_or(0) + pub async fn remove_peer(&self, peer: PeerId) -> Result { + self.ask(|reply| RibCommand::RemovePeer { peer, reply }) + .await } /// Returns a point-in-time snapshot of all routes in the RIB. - pub async fn snapshot(&self) -> Vec { - let (reply_tx, reply_rx) = oneshot::channel(); - let _ = self.tx.send(RibCommand::Snapshot { reply: reply_tx }).await; - reply_rx.await.unwrap_or_default() + pub async fn snapshot(&self) -> Result, RibStopped> { + self.ask(|reply| RibCommand::Snapshot { reply }).await + } + + /// How many routes the RIB holds, without copying them. + pub async fn len(&self) -> Result { + self.ask(|reply| RibCommand::Len { reply }).await + } + + /// Whether the RIB holds no routes. + pub async fn is_empty(&self) -> Result { + Ok(self.len().await? == 0) + } + + /// Sends a command that is answered, and waits for the answer. + async fn ask( + &self, + command: impl FnOnce(oneshot::Sender) -> RibCommand, + ) -> Result { + let (reply, answer) = oneshot::channel(); + self.tx.send(command(reply)).await.map_err(|_| STOPPED)?; + answer.await.map_err(|_| STOPPED) + } + + /// Resolves when the actor has stopped. It never does in a healthy + /// process; whoever owns the RIB waits on this to notice a failure even + /// when nothing happens to be using the table. + pub async fn stopped(&self) { + self.tx.closed().await + } + + /// Stops the actor, as a panic in it would. For tests of how the rest + /// of the server copes. + #[doc(hidden)] + pub fn stop_for_testing(&self) { + self.actor.abort(); } /// Subscribe to a stream of RIB mutation events. @@ -132,9 +173,11 @@ async fn run_actor(mut rx: mpsc::Receiver, events: broadcast::Sender }) }) .collect(); - if reply.send(snap).is_err() { - warn!("Snapshot reply channel closed before send"); - } + // A failed reply means the asker stopped waiting. + let _ = reply.send(snap); + } + RibCommand::Len { reply } => { + let _ = reply.send(table.values().map(HashMap::len).sum()); } } } @@ -177,29 +220,32 @@ mod tests { async fn a_prefix_withdrawn_and_announced_by_one_update_ends_up_announced() { let rib = AdjRibIn::spawn(); let a = peer(179, 65001); - rib.apply(a, announce(&["10.0.0.0/8"])).await; + rib.apply(a, announce(&["10.0.0.0/8"])).await.unwrap(); let mut update = announce(&["10.0.0.0/8"]); update.withdrawn = vec![prefix("10.0.0.0/8")]; - rib.apply(a, update).await; + rib.apply(a, update).await.unwrap(); - assert_eq!(rib.snapshot().await.len(), 1); + assert_eq!(rib.snapshot().await.unwrap().len(), 1); } #[tokio::test] async fn removing_a_peer_leaves_other_peers_routes_alone() { let rib = AdjRibIn::spawn(); let (a, b) = (peer(179, 65001), peer(180, 65002)); - rib.apply(a, announce(&["10.0.0.0/8", "10.1.0.0/16"])).await; - rib.apply(b, announce(&["10.0.0.0/8"])).await; + rib.apply(a, announce(&["10.0.0.0/8", "10.1.0.0/16"])) + .await + .unwrap(); + rib.apply(b, announce(&["10.0.0.0/8"])).await.unwrap(); // Commands are handled in order, so once the snapshot is back the // announcements above have been applied and their events sent. - assert_eq!(rib.snapshot().await.len(), 3); + assert_eq!(rib.snapshot().await.unwrap().len(), 3); + assert_eq!(rib.len().await, Ok(3)); let mut events = rib.subscribe(); - assert_eq!(rib.remove_peer(a).await, 2); + assert_eq!(rib.remove_peer(a).await, Ok(2)); - let left = rib.snapshot().await; + let left = rib.snapshot().await.unwrap(); assert_eq!(left.len(), 1); assert_eq!(left[0].peer, b); for _ in 0..2 { @@ -207,6 +253,25 @@ mod tests { matches!(events.recv().await.unwrap(), RibEvent::Withdrawn { peer, .. } if peer == a) ); } - assert_eq!(rib.remove_peer(a).await, 0, "nothing left to remove"); + assert_eq!(rib.remove_peer(a).await, Ok(0), "nothing left to remove"); + } + + /// A stopped actor used to look like an empty table: updates vanished, + /// and a snapshot came back with no routes. + #[tokio::test] + async fn a_stopped_actor_is_an_error_and_not_an_empty_table() { + let rib = AdjRibIn::spawn(); + let a = peer(179, 65001); + rib.apply(a, announce(&["10.0.0.0/8"])).await.unwrap(); + assert_eq!(rib.len().await, Ok(1)); + + rib.stop_for_testing(); + rib.stopped().await; + + assert_eq!(rib.apply(a, announce(&["10.1.0.0/16"])).await, Err(STOPPED)); + assert_eq!(rib.snapshot().await.map(|s| s.len()), Err(STOPPED)); + assert_eq!(rib.len().await, Err(STOPPED)); + assert_eq!(rib.remove_peer(a).await, Err(STOPPED)); + assert_eq!(STOPPED.to_string(), "the Adj-RIB-In has stopped"); } } diff --git a/crates/rib/src/error.rs b/crates/rib/src/error.rs new file mode 100644 index 0000000..71982c6 --- /dev/null +++ b/crates/rib/src/error.rs @@ -0,0 +1,14 @@ +/// A RIB's actor is no longer running, so the table can be neither read nor +/// changed. +/// +/// The actor runs for as long as a handle to it exists; it stops early only +/// by panicking. That is a bug, and one the server cannot work around: +/// routes would be accepted from peers and go nowhere. Callers pass this up +/// until the process exits and is restarted, rather than carry on with a +/// table that looks empty. +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +#[error("the {rib} has stopped")] +pub struct RibStopped { + /// Which table: "Adj-RIB-In" or "local RIB". + pub rib: &'static str, +} diff --git a/crates/rib/src/lib.rs b/crates/rib/src/lib.rs index c64cddd..2e18adc 100644 --- a/crates/rib/src/lib.rs +++ b/crates/rib/src/lib.rs @@ -1,9 +1,11 @@ mod adj_rib_in; mod entry; +mod error; mod event; mod local_rib; pub use adj_rib_in::AdjRibIn; pub use entry::{PeerId, ReceivedUpdate, RibEntry, SnapshotEntry}; +pub use error::RibStopped; pub use event::RibEvent; -pub use local_rib::{LocalRib, LocalRibError, LocalRibEvent, LocalRoute, LocalRouteAttrs}; +pub use local_rib::{LocalRib, LocalRibEvent, LocalRoute, LocalRouteAttrs}; diff --git a/crates/rib/src/local_rib.rs b/crates/rib/src/local_rib.rs index e003ed2..b1f15dc 100644 --- a/crates/rib/src/local_rib.rs +++ b/crates/rib/src/local_rib.rs @@ -12,7 +12,9 @@ use bgp_packet::message::{LargeCommunity, OriginPathAttribute}; use bgp_primitives::ip_prefix::IpPrefix; use serde::{Deserialize, Serialize}; use tokio::sync::{broadcast, mpsc, oneshot}; -use tracing::warn; +use tokio::task::AbortHandle; + +use crate::error::RibStopped; const EVENT_CHANNEL_CAPACITY: usize = 1024; const CMD_CHANNEL_CAPACITY: usize = 256; @@ -61,11 +63,7 @@ impl LocalRoute { } } -#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] -pub enum LocalRibError { - #[error("local RIB actor has stopped")] - Closed, -} +const STOPPED: RibStopped = RibStopped { rib: "local RIB" }; #[derive(Debug, Clone)] pub enum LocalRibEvent { @@ -88,49 +86,76 @@ enum Command { Snapshot { reply: oneshot::Sender>, }, + Len { + reply: oneshot::Sender, + }, } /// Handle to the local RIB actor. Cheap to clone; all clones share one table. +/// +/// Every operation fails with [`RibStopped`] if the actor is gone; see +/// there for what that means. #[derive(Clone)] pub struct LocalRib { tx: mpsc::Sender, events: broadcast::Sender, + actor: AbortHandle, } impl LocalRib { pub fn spawn() -> Self { let (tx, rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); - tokio::spawn(run_actor(rx, events.clone())); - Self { tx, events } + let actor = tokio::spawn(run_actor(rx, events.clone())).abort_handle(); + Self { tx, events, actor } } /// Adds a route, or replaces the existing route for the same prefix. - pub async fn announce(&self, route: LocalRoute) -> Result<(), LocalRibError> { - let (reply, rx) = oneshot::channel(); - self.tx - .send(Command::Announce { route, reply }) - .await - .map_err(|_| LocalRibError::Closed)?; - rx.await.map_err(|_| LocalRibError::Closed) + pub async fn announce(&self, route: LocalRoute) -> Result<(), RibStopped> { + self.ask(|reply| Command::Announce { route, reply }).await } /// Removes a route. Returns the removed route, or `None` if the prefix /// was not present. - pub async fn withdraw(&self, prefix: IpPrefix) -> Result, LocalRibError> { - let (reply, rx) = oneshot::channel(); - self.tx - .send(Command::Withdraw { prefix, reply }) - .await - .map_err(|_| LocalRibError::Closed)?; - rx.await.map_err(|_| LocalRibError::Closed) + pub async fn withdraw(&self, prefix: IpPrefix) -> Result, RibStopped> { + self.ask(|reply| Command::Withdraw { prefix, reply }).await } /// Point-in-time copy of every route, in no particular order. - pub async fn snapshot(&self) -> Vec { - let (reply, rx) = oneshot::channel(); - let _ = self.tx.send(Command::Snapshot { reply }).await; - rx.await.unwrap_or_default() + pub async fn snapshot(&self) -> Result, RibStopped> { + self.ask(|reply| Command::Snapshot { reply }).await + } + + /// How many routes the RIB holds, without copying them. + pub async fn len(&self) -> Result { + self.ask(|reply| Command::Len { reply }).await + } + + /// Whether the RIB holds no routes. + pub async fn is_empty(&self) -> Result { + Ok(self.len().await? == 0) + } + + /// Sends a command and waits for its answer. + async fn ask( + &self, + command: impl FnOnce(oneshot::Sender) -> Command, + ) -> Result { + let (reply, answer) = oneshot::channel(); + self.tx.send(command(reply)).await.map_err(|_| STOPPED)?; + answer.await.map_err(|_| STOPPED) + } + + /// Resolves when the actor has stopped. See [`crate::AdjRibIn::stopped`]. + pub async fn stopped(&self) { + self.tx.closed().await + } + + /// Stops the actor, as a panic in it would. For tests of how the rest + /// of the server copes. + #[doc(hidden)] + pub fn stop_for_testing(&self) { + self.actor.abort(); } /// Subscribe to mutations. See [`crate::AdjRibIn::subscribe`] for lag @@ -158,9 +183,11 @@ async fn run_actor(mut rx: mpsc::Receiver, events: broadcast::Sender { - if reply.send(table.values().cloned().collect()).is_err() { - warn!("Snapshot reply channel closed before send"); - } + // A failed reply means the asker stopped waiting. + let _ = reply.send(table.values().cloned().collect()); + } + Command::Len { reply } => { + let _ = reply.send(table.len()); } } } @@ -187,14 +214,14 @@ mod tests { assert!( matches!(events.recv().await.unwrap(), LocalRibEvent::Announced(r) if r.prefix == prefix("10.0.0.0/24")) ); - assert_eq!(rib.snapshot().await.len(), 1); + assert_eq!(rib.len().await, Ok(1)); let removed = rib.withdraw(prefix("10.0.0.0/24")).await.unwrap(); assert_eq!(removed.unwrap().prefix, prefix("10.0.0.0/24")); assert!( matches!(events.recv().await.unwrap(), LocalRibEvent::Withdrawn { prefix: p } if p == prefix("10.0.0.0/24")) ); - assert!(rib.snapshot().await.is_empty()); + assert_eq!(rib.is_empty().await, Ok(true)); // Withdrawing an absent prefix is a no-op, not an error. assert_eq!(rib.withdraw(prefix("10.0.0.0/24")).await.unwrap(), None); @@ -212,8 +239,22 @@ mod tests { .await .unwrap(); - let table = rib.snapshot().await; + let table = rib.snapshot().await.unwrap(); assert_eq!(table.len(), 1); assert_eq!(table[0].attrs, attrs); } + + #[tokio::test] + async fn a_stopped_actor_is_an_error_and_not_an_empty_table() { + let rib = LocalRib::spawn(); + rib.announce(route("10.0.0.0/24")).await.unwrap(); + + rib.stop_for_testing(); + rib.stopped().await; + + assert_eq!(rib.announce(route("10.0.1.0/24")).await, Err(STOPPED)); + assert_eq!(rib.withdraw(prefix("10.0.0.0/24")).await, Err(STOPPED)); + assert_eq!(rib.snapshot().await, Err(STOPPED)); + assert_eq!(rib.len().await, Err(STOPPED)); + } } diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 25ff71f..3cbd491 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -16,7 +16,7 @@ use bgp_packet::message::{ }; use bgp_primitives::constants::{AddressFamilyId, Afi, Safi}; use bgp_primitives::ip_prefix::IpPrefix; -use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute}; +use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute, RibStopped}; use tokio::sync::broadcast::error::RecvError; use tokio::task::JoinHandle; use tracing::{debug, info, warn}; @@ -31,6 +31,8 @@ pub fn spawn_advertiser( tokio::spawn(run(local_rib, registry, local_asn)) } +/// Runs until the local RIB stops. The server watches for this task ending: +/// without it nothing is announced. async fn run(local_rib: LocalRib, registry: SessionRegistry, local_asn: u32) { let mut rib_events = local_rib.subscribe(); let mut session_events = registry.subscribe(); @@ -50,20 +52,22 @@ async fn run(local_rib: LocalRib, registry: SessionRegistry, local_asn: u32) { } Err(RecvError::Lagged(n)) => { warn!(missed = n, "Advertiser lagged on local RIB events, resyncing all sessions"); - resync_all(&local_rib, ®istry, local_asn).await; + if resync_all(&local_rib, ®istry, local_asn).await.is_err() { + return; + } } Err(RecvError::Closed) => return, }, ev = session_events.recv() => match ev { Ok(SessionEvent::Up(session)) => { - let routes = local_rib.snapshot().await; + let Ok(routes) = local_rib.snapshot().await else { return }; info!(peer = %session.peer, routes = routes.len(), "Session up, sending local routes"); for route in &routes { send_route(&session, route, local_asn).await; } } Ok(SessionEvent::RefreshRequested(session)) => { - let routes = local_rib.snapshot().await; + let Ok(routes) = local_rib.snapshot().await else { return }; 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; @@ -72,7 +76,9 @@ async fn run(local_rib: LocalRib, registry: SessionRegistry, local_asn: u32) { Ok(SessionEvent::Down(_)) => {} Err(RecvError::Lagged(n)) => { warn!(missed = n, "Advertiser lagged on session events, resyncing all sessions"); - resync_all(&local_rib, ®istry, local_asn).await; + if resync_all(&local_rib, ®istry, local_asn).await.is_err() { + return; + } } Err(RecvError::Closed) => return, }, @@ -80,13 +86,18 @@ async fn run(local_rib: LocalRib, registry: SessionRegistry, local_asn: u32) { } } -async fn resync_all(local_rib: &LocalRib, registry: &SessionRegistry, local_asn: u32) { - let routes = local_rib.snapshot().await; +async fn resync_all( + local_rib: &LocalRib, + registry: &SessionRegistry, + local_asn: u32, +) -> Result<(), RibStopped> { + let routes = local_rib.snapshot().await?; for session in registry.list() { for route in &routes { send_route(&session, route, local_asn).await; } } + Ok(()) } async fn send_route(session: &SessionHandle, route: &LocalRoute, local_asn: u32) { diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index ba1e974..5953231 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -16,14 +16,14 @@ use bgp_packet::errors::{BgpError, DecodeError, MsgHeaderSubcode}; use bgp_packet::message::{BGP_HEADER_LEN, BGP_MAX_MSG_LEN, Message, PathAttribute, UpdateMessage}; use bgp_packet::update_decode::UpdateError; use bgp_primitives::parser::ParserContext; -use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate}; +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::sync::mpsc; use tokio::time::{Instant, Interval, Sleep, interval_at, sleep}; -use tracing::{info, warn}; +use tracing::{error, info, warn}; use crate::collision::{CollisionTable, ConnectionGuard, Direction}; use crate::error::SessionError; @@ -105,12 +105,17 @@ pub async fn run_session( // (RFC 4271 §8.2.2). Do this before the registration drops so // subscribers see the withdrawals before the Down event. if let Some(established) = session.established.take() { - let removed = session - .config - .adj_rib_in - .remove_peer(established.peer_id) - .await; - info!(peer = %established.peer_id, removed_routes = removed, "Session ended"); + let Established { + peer_id, + routes, + _registration, + .. + } = established; + match routes.remove().await { + Ok(removed) => info!(peer = %peer_id, removed_routes = removed, "Session ended"), + // No table left to remove them from. + Err(e) => error!(peer = %peer_id, err = %e, "Session ended"), + } } result } @@ -120,10 +125,66 @@ struct Established { peer_id: PeerId, /// Messages the rest of the server queued for the peer. outbound: mpsc::Receiver, + /// The routes learned from the peer, which go when the session does. + routes: PeerRoutes, /// Keeps the session listed in the registry. _registration: SessionGuard, } +/// The routes a peer has announced on a session, as far as their lifetime +/// goes: they must not outlive it (RFC 4271 §8.2.2). +/// +/// A session that ends removes them with [`PeerRoutes::remove`] and waits, +/// so that their withdrawal is seen before the session is seen to be down. +/// If the session's task is torn down without getting there (a panic, or +/// the task being aborted), dropping this removes them all the same. Left +/// in the table they would be offered as routes of a peer that is gone. +struct PeerRoutes { + adj_rib_in: AdjRibIn, + peer: PeerId, + removed: bool, +} + +impl PeerRoutes { + fn new(adj_rib_in: AdjRibIn, peer: PeerId) -> Self { + Self { + adj_rib_in, + peer, + removed: false, + } + } + + /// Removes the routes and returns how many there were. + async fn remove(mut self) -> Result { + self.removed = true; + self.adj_rib_in.remove_peer(self.peer).await + } +} + +impl Drop for PeerRoutes { + fn drop(&mut self) { + if self.removed { + return; + } + // Not reached in the normal course of things. Drop cannot wait, so + // the removal is handed to a task, if there is still a runtime to + // run it. + let (adj_rib_in, peer) = (self.adj_rib_in.clone(), self.peer); + if let Ok(runtime) = tokio::runtime::Handle::try_current() { + runtime.spawn(async move { + match adj_rib_in.remove_peer(peer).await { + Ok(removed) => warn!( + %peer, + removed_routes = removed, + "Session task ended abnormally; removed its routes" + ), + Err(e) => error!(%peer, err = %e, "Could not remove routes of ended session"), + } + }); + } + } +} + struct Session { fsm: Fsm, frames: FrameReader, @@ -257,10 +318,16 @@ impl Session { if let Some(established) = &self.established { let (withdrawn, announced) = prefix_counts(&update); info!(peer = %established.peer_id, withdrawn, announced, "Received UPDATE"); - self.config + let applied = self + .config .adj_rib_in .apply(established.peer_id, ReceivedUpdate::from_message(&update)) .await; + if let Err(e) = applied { + // What the machine wanted next no longer applies. + error!(peer = %established.peer_id, err = %e, "Cannot store routes"); + queue = self.fsm.handle(Event::RibStopped(e)).into(); + } } } Action::RouteRefreshRequested => { @@ -300,6 +367,7 @@ impl Session { self.established = Some(Established { peer_id, outbound, + routes: PeerRoutes::new(self.config.adj_rib_in.clone(), peer_id), _registration: registration, }); } @@ -528,6 +596,40 @@ mod tests { )); } + /// A session task that is torn down without running its cleanup must + /// not leave the peer's routes behind. + #[tokio::test] + async fn routes_are_removed_even_if_the_session_never_cleans_up() { + let adj_rib_in = AdjRibIn::spawn(); + let peer = PeerId { + addr: ([192, 0, 2, 1], 179).into(), + asn: 65001, + }; + let update = UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![], + announced_routes: vec![ + bgp_primitives::ip_prefix::IpPrefix::try_from("10.0.0.0/8").unwrap(), + ], + }; + adj_rib_in + .apply(peer, ReceivedUpdate::from_message(&update)) + .await + .unwrap(); + assert_eq!(adj_rib_in.len().await, Ok(1)); + + // As when a panic unwinds the session's task. + drop(PeerRoutes::new(adj_rib_in.clone(), peer)); + + tokio::time::timeout(Duration::from_secs(5), async { + while adj_rib_in.len().await != Ok(0) { + tokio::task::yield_now().await; + } + }) + .await + .expect("the routes are removed"); + } + proptest! { /// TCP delivers a stream, not messages: however the bytes of a run /// of messages are cut into reads, the same messages come out. diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index 03be11f..17b9924 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -6,6 +6,7 @@ use std::net::SocketAddr; use bgp_packet::errors::{BgpError, CeaseSubcode, DecodeError, FsmSubcode, OpenMsgSubcode}; use bgp_packet::message::NotificationMessage; use bgp_primitives::parser::EncodeError; +use bgp_rib::RibStopped; use thiserror::Error; use crate::family::FamilySet; @@ -41,6 +42,10 @@ pub enum SessionError { Collision, /// The peer offers none of the address families configured for it, so /// the session could carry nothing. + /// The table the session's routes go into is gone. The server is on + /// its way down; the peer is told so. + #[error(transparent)] + RibStopped(#[from] RibStopped), #[error("no address family in common: we offer {ours}, the peer offers {theirs}")] NoCommonAddressFamily { ours: FamilySet, theirs: FamilySet }, } @@ -68,6 +73,7 @@ impl SessionError { Self::OpenRefused(error) => error.clone(), Self::WrongPeerAs { .. } => BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), Self::Collision => BgpError::Cease(CeaseSubcode::ConnCollisionResolution), + Self::RibStopped(_) => BgpError::Cease(CeaseSubcode::AdminShutdown), // RFC 5492 §5: the Data lists the capabilities we need and the // peer did not offer. Self::NoCommonAddressFamily { ours, .. } => { @@ -95,8 +101,13 @@ pub enum ServerError { }, #[error("listener failed: {0}")] Io(#[from] io::Error), - #[error("the local RIB stopped during startup")] - LocalRibStopped, + /// A RIB's actor died. The server cannot work without its tables. + #[error(transparent)] + RibStopped(#[from] RibStopped), + /// A task the server depends on ended, which none of them do unless + /// they panic. + #[error("the {task} task stopped unexpectedly: {reason}")] + TaskStopped { task: &'static str, reason: String }, } /// Why a configuration was rejected. diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index ec0b675..84f6aa1 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -18,6 +18,8 @@ use bgp_packet::errors::{BgpError, DecodeError, OpenMsgSubcode}; use bgp_packet::message::{AS_TRANS, BGP4_VERSION, Message, UpdateMessage}; use bgp_packet::open::{Capability, OpenMessage, OpenOption}; +use bgp_rib::RibStopped; + use crate::error::SessionError; use crate::family::FamilySet; use crate::metrics::message_type_name; @@ -92,6 +94,9 @@ pub enum Event { CollisionLost, /// The rest of the server wants this message sent to the peer. Outbound(Message), + /// The table that routes from the peer go into has stopped, so the + /// session cannot go on. + RibStopped(RibStopped), } /// Something the driver must do, in the order given. @@ -180,6 +185,7 @@ impl Fsm { Event::Message(msg) => self.on_message(msg), Event::DecodeFailed(e) => self.fail(SessionError::Decode(e)), Event::CollisionLost => self.fail(SessionError::Collision), + Event::RibStopped(e) => self.fail(SessionError::RibStopped(e)), Event::HoldTimerExpired => self.fail(match self.phase { Phase::Established(_) => SessionError::HoldTimerExpired, _ => SessionError::Timeout(self.state()), @@ -835,6 +841,16 @@ mod tests { assert_eq!(fsm.state(), State::Closed); } + #[test] + fn losing_the_rib_ends_the_session_and_tells_the_peer_we_are_going() { + let mut fsm = fsm_in(State::Established, 90); + let stopped = RibStopped { rib: "Adj-RIB-In" }; + let actions = fsm.handle(Event::RibStopped(stopped)); + let (sent, result) = ending(&fsm, actions); + assert_eq!(sent, Some(BgpError::Cease(CeaseSubcode::AdminShutdown))); + assert!(matches!(result, Err(SessionError::RibStopped(e)) if e == stopped)); + } + #[test] fn a_closed_machine_ignores_everything() { let mut fsm = fsm_in(State::Established, 90); diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index b00e5c5..164daf7 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -1,12 +1,15 @@ use std::io; use std::net::SocketAddr; +use std::pin::Pin; use std::sync::Arc; +use std::time::Duration; use futures::stream::{FuturesUnordered, StreamExt}; use tokio::net::{TcpListener, TcpStream}; -use tracing::{error, info}; +use tokio::task::{JoinError, JoinHandle}; +use tracing::{error, info, warn}; -use bgp_rib::{AdjRibIn, LocalRib}; +use bgp_rib::{AdjRibIn, LocalRib, RibStopped}; use crate::advertise::spawn_advertiser; use crate::collision::{CollisionTable, Direction}; @@ -57,10 +60,7 @@ impl BgpServerBuilder { // each session is sent the whole table when it comes up. let local_rib = LocalRib::spawn(); for route in &self.config.originate { - local_rib - .announce(route.clone()) - .await - .map_err(|_| ServerError::LocalRibStopped)?; + local_rib.announce(route.clone()).await?; } Ok(BgpServer { listeners, @@ -119,16 +119,24 @@ impl BgpServer { families: FamilySet::supported(), }; - let _advertiser = spawn_advertiser( - self.local_rib.clone(), - self.registry.clone(), - self.config.local_asn, - ); - let _dialers = spawn_dialers( + // The tasks the server cannot do without. None of them ends while + // the server runs, unless it panics. + let mut tasks: FuturesUnordered<_> = spawn_dialers( &self.config.peers, &session_config, self.config.connect_retry, - ); + ) + .into_iter() + .map(|dialer| named("dialer", dialer)) + .collect(); + tasks.push(named( + "advertiser", + spawn_advertiser( + self.local_rib.clone(), + self.registry.clone(), + self.config.local_asn, + ), + )); let mut accepting: FuturesUnordered<_> = self .listeners @@ -136,7 +144,13 @@ impl BgpServer { .map(|(addr, listener)| accept_one(addr, listener)) .collect(); - while let Some((listen_addr, result, listener)) = accepting.next().await { + loop { + let (listen_addr, result, listener) = tokio::select! { + Some(accepted) = accepting.next() => accepted, + failure = failed(&self.adj_rib_in, &self.local_rib, &mut tasks) => { + return Err(give_up(&self.registry, failure).await); + } + }; match result { Ok((stream, peer_addr)) => { // Inbound connections from a configured peer inherit its @@ -161,8 +175,52 @@ impl BgpServer { } accepting.push(accept_one(listen_addr, listener)); } + } +} - Ok(()) +/// Ends the server because a part of it that it cannot do without has +/// failed. Peers are told we are going (Cease / Administrative Shutdown), +/// so that they stop using our routes now and not when their hold timers +/// expire. +async fn give_up(registry: &SessionRegistry, failure: ServerError) -> ServerError { + error!(err = %failure, "Shutting down: the server cannot continue"); + let left = registry.shutdown(SHUTDOWN_GRACE).await; + if left > 0 { + warn!(sessions = left, "Some sessions did not close in time"); + } + failure +} + +/// How long sessions get to send their Cease when the server gives up. +const SHUTDOWN_GRACE: Duration = Duration::from_secs(2); + +type NamedTask = Pin)> + Send>>; + +fn named(name: &'static str, task: JoinHandle<()>) -> NamedTask { + Box::pin(async move { (name, task.await) }) +} + +/// Resolves when something the server depends on has stopped: a RIB's +/// actor or one of `tasks`. In a healthy server it never resolves. +/// +/// Nothing else would notice. A stopped RIB is only found out by whoever +/// next uses it, and a stopped task by nobody at all: the server would go +/// on accepting sessions while storing or announcing nothing. +async fn failed( + adj_rib_in: &AdjRibIn, + local_rib: &LocalRib, + tasks: &mut FuturesUnordered, +) -> ServerError { + tokio::select! { + () = adj_rib_in.stopped() => RibStopped { rib: "Adj-RIB-In" }.into(), + () = local_rib.stopped() => RibStopped { rib: "local RIB" }.into(), + Some((task, result)) = tasks.next() => ServerError::TaskStopped { + task, + reason: match result { + Ok(()) => "it returned".into(), + Err(e) => e.to_string(), + }, + }, } } diff --git a/crates/web/src/api.rs b/crates/web/src/api.rs index c2b9436..89c944b 100644 --- a/crates/web/src/api.rs +++ b/crates/web/src/api.rs @@ -5,9 +5,11 @@ use std::time::{SystemTime, UNIX_EPOCH}; use axum::Json; use axum::extract::State; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; use bgp_packet::community; use bgp_packet::message::{OriginPathAttribute, PathAttribute}; -use bgp_rib::{LocalRoute, RibEntry, SnapshotEntry}; +use bgp_rib::{LocalRoute, RibEntry, RibStopped, SnapshotEntry}; use bgp_server::session::{ConfiguredPeer, DialState, SessionHandle, SessionRegistry}; use serde::Serialize; @@ -258,8 +260,24 @@ fn origin_name(o: OriginPathAttribute) -> &'static str { // Handlers // --------------------------------------------------------------------------- -pub async fn status(State(state): State) -> Json { - Json(StatusView { +/// A table could not be read because its actor has stopped. The daemon is +/// on its way down; saying so beats answering with an empty table. +pub struct Unavailable(RibStopped); + +impl From for Unavailable { + fn from(e: RibStopped) -> Self { + Self(e) + } +} + +impl IntoResponse for Unavailable { + fn into_response(self) -> Response { + (StatusCode::SERVICE_UNAVAILABLE, self.0.to_string()).into_response() + } +} + +pub async fn status(State(state): State) -> Result, Unavailable> { + Ok(Json(StatusView { router_id: state.info.router_id.to_string(), local_asn: state.info.local_asn, version: state.info.version.clone(), @@ -267,9 +285,9 @@ pub async fn status(State(state): State) -> Json { started_at_unix: unix_secs(state.info.started_at), uptime_secs: state.info.started.elapsed().as_secs(), sessions: state.registry.list().len(), - local_routes: state.local_rib.snapshot().await.len(), - received_routes: state.adj_rib_in.snapshot().await.len(), - }) + local_routes: state.local_rib.len().await?, + received_routes: state.adj_rib_in.len().await?, + })) } /// The configuration file the daemon was started with. @@ -290,26 +308,30 @@ pub async fn peers(State(state): State) -> Json> { Json(peer_views(&state.registry)) } -pub async fn local_routes(State(state): State) -> Json> { +pub async fn local_routes( + State(state): State, +) -> Result>, Unavailable> { let mut routes: Vec = state .local_rib .snapshot() - .await + .await? .iter() .map(LocalRouteView::from) .collect(); routes.sort_by(|a, b| a.prefix.cmp(&b.prefix)); - Json(routes) + Ok(Json(routes)) } -pub async fn received_routes(State(state): State) -> Json> { +pub async fn received_routes( + State(state): State, +) -> Result>, Unavailable> { let mut routes: Vec = state .adj_rib_in .snapshot() - .await + .await? .iter() .map(ReceivedRouteView::from) .collect(); routes.sort_by(|a, b| (&a.prefix, &a.peer_addr).cmp(&(&b.prefix, &b.peer_addr))); - Json(routes) + Ok(Json(routes)) } diff --git a/crates/web/src/metrics.rs b/crates/web/src/metrics.rs index 4f7636f..bce368e 100644 --- a/crates/web/src/metrics.rs +++ b/crates/web/src/metrics.rs @@ -4,14 +4,26 @@ use axum::extract::State; use axum::http::{StatusCode, header}; use axum::response::IntoResponse; +use bgp_rib::RibStopped; use bgp_server::metrics::{ADJ_RIB_IN_ROUTES, LOCAL_ROUTES}; use prometheus::{Encoder, TextEncoder}; use crate::AppState; pub async fn metricz(State(state): State) -> impl IntoResponse { - LOCAL_ROUTES.set(state.local_rib.snapshot().await.len() as i64); - ADJ_RIB_IN_ROUTES.set(state.adj_rib_in.snapshot().await.len() as i64); + // A scrape that cannot read the tables fails. Serving the last known + // sizes would hide that the daemon is broken from the one system whose + // job is to notice. + let sizes = async { + Ok::<_, RibStopped>((state.local_rib.len().await?, state.adj_rib_in.len().await?)) + }; + match sizes.await { + Ok((local, received)) => { + LOCAL_ROUTES.set(i64::try_from(local).unwrap_or(i64::MAX)); + ADJ_RIB_IN_ROUTES.set(i64::try_from(received).unwrap_or(i64::MAX)); + } + Err(e) => return (StatusCode::SERVICE_UNAVAILABLE, e.to_string()).into_response(), + } let mut buf = Vec::new(); match TextEncoder::new().encode(&prometheus::gather(), &mut buf) { diff --git a/crates/web/static/statusz.html b/crates/web/static/statusz.html index 3eb0f4d..c86db4f 100644 --- a/crates/web/static/statusz.html +++ b/crates/web/static/statusz.html @@ -120,7 +120,7 @@ View: variables, status json, < async function getJSON(url) { const r = await fetch(url); - if (!r.ok) throw new Error(`${url}: ${r.status}`); + if (!r.ok) throw new Error(`${url}: ${r.status} ${await r.text()}`); return r.json(); }