From 2ab6c1c38bb8419453b1460cf0fdd63c4426bf66 Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Thu, 1 Oct 2026 20:58:12 +0200 Subject: [PATCH] Format the workspace with rustfmt Default settings. No change in behaviour; the tree was last formatted before the lecture work and had drifted to a mix of line widths. Co-Authored-By: Claude Fable 5.1 --- .../integration-tests/examples/fake_peer.rs | 38 ++++- crates/integration-tests/src/test_peer.rs | 15 +- crates/integration-tests/src/test_server.rs | 9 +- crates/integration-tests/tests/announce.rs | 145 +++++++++++++----- .../tests/basic_connection.rs | 12 +- crates/integration-tests/tests/collision.rs | 41 ++++- crates/integration-tests/tests/dialer.rs | 64 ++++++-- crates/integration-tests/tests/http_api.rs | 95 ++++++++++-- crates/integration-tests/tests/interop.rs | 35 ++++- crates/packet/src/community.rs | 24 ++- crates/packet/src/open.rs | 16 +- crates/rib/src/adj_rib_in.rs | 61 ++++++-- crates/rib/src/entry.rs | 5 +- crates/rib/src/event.rs | 11 +- crates/rib/src/local_rib.rs | 41 +++-- crates/server/src/advertise.rs | 64 ++++++-- crates/server/src/collision.rs | 76 +++++++-- crates/server/src/config.rs | 61 ++++++-- crates/server/src/dialer.rs | 19 ++- crates/server/src/fsm.rs | 73 +++++++-- crates/server/src/metrics.rs | 40 +++-- crates/server/src/server.rs | 24 ++- crates/server/src/session.rs | 66 ++++++-- crates/web/src/api.rs | 70 +++++++-- crates/web/src/events.rs | 91 ++++++++--- crates/web/src/lib.rs | 5 +- crates/web/src/metrics.rs | 7 +- 27 files changed, 959 insertions(+), 249 deletions(-) diff --git a/crates/integration-tests/examples/fake_peer.rs b/crates/integration-tests/examples/fake_peer.rs index b394803..a375878 100644 --- a/crates/integration-tests/examples/fake_peer.rs +++ b/crates/integration-tests/examples/fake_peer.rs @@ -78,8 +78,13 @@ async fn main() -> Result<()> { } async fn run(args: &Args, mut peer: TestPeer, addr: std::net::SocketAddr) -> Result<()> { - let open = peer.handshake(args.asn, args.router_id, args.hold_time).await?; - println!("established with {} (AS{} router-id {})", addr, open.asn, open.identifier); + let open = peer + .handshake(args.asn, args.router_id, args.hold_time) + .await?; + println!( + "established with {} (AS{} router-id {})", + addr, open.asn, open.identifier + ); for p in &args.announce { let prefix = IpPrefix::try_from(p.as_str())?; @@ -123,9 +128,15 @@ fn announce(prefix: IpPrefix, as_path: Vec, communities: Vec<(u16, u16)>) - PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(10, 0, 0, 2))), ]; if !communities.is_empty() { - path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute(communities))); + path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute( + communities, + ))); + } + UpdateMessage { + withdrawn_routes: vec![], + path_attributes, + announced_routes: vec![prefix], } - UpdateMessage { withdrawn_routes: vec![], path_attributes, announced_routes: vec![prefix] } } async fn echo_back(peer: &mut TestPeer, u: &UpdateMessage, asn: u32) -> Result<()> { @@ -133,7 +144,9 @@ async fn echo_back(peer: &mut TestPeer, u: &UpdateMessage, asn: u32) -> Result<( let mut communities = vec![]; for attr in &u.path_attributes { match attr { - PathAttribute::AsPath(p) => as_path.extend(p.segments.iter().flat_map(|s| s.path.iter().copied())), + PathAttribute::AsPath(p) => { + as_path.extend(p.segments.iter().flat_map(|s| s.path.iter().copied())) + } PathAttribute::Communities(c) => communities = c.0.clone(), _ => {} } @@ -143,7 +156,12 @@ async fn echo_back(peer: &mut TestPeer, u: &UpdateMessage, asn: u32) -> Result<( return Ok(()); } for prefix in &u.announced_routes { - peer.send(&Message::Update(announce(prefix.clone(), as_path.clone(), communities.clone()))).await?; + peer.send(&Message::Update(announce( + prefix.clone(), + as_path.clone(), + communities.clone(), + ))) + .await?; println!(" echoed {prefix} back with path {as_path:?}"); } if !u.withdrawn_routes.is_empty() { @@ -153,7 +171,13 @@ async fn echo_back(peer: &mut TestPeer, u: &UpdateMessage, asn: u32) -> Result<( announced_routes: vec![], })) .await?; - println!(" echoed withdraw of {:?}", u.withdrawn_routes.iter().map(|p| p.to_string()).collect::>()); + println!( + " echoed withdraw of {:?}", + u.withdrawn_routes + .iter() + .map(|p| p.to_string()) + .collect::>() + ); } Ok(()) } diff --git a/crates/integration-tests/src/test_peer.rs b/crates/integration-tests/src/test_peer.rs index ad19ed6..e107151 100644 --- a/crates/integration-tests/src/test_peer.rs +++ b/crates/integration-tests/src/test_peer.rs @@ -25,7 +25,10 @@ impl TestPeer { /// Wraps an already-connected stream, e.g. one accepted from a server /// that dialled us. pub fn from_stream(stream: TcpStream) -> Self { - Self { stream, ctx: ParserContext::default() } + Self { + stream, + ctx: ParserContext::default(), + } } /// Send a BGP message. @@ -48,7 +51,9 @@ impl TestPeer { let mut buf = vec![0u8; total_len]; buf[..BGP_HEADER_LEN as usize].copy_from_slice(&header); - self.stream.read_exact(&mut buf[BGP_HEADER_LEN as usize..]).await?; + self.stream + .read_exact(&mut buf[BGP_HEADER_LEN as usize..]) + .await?; let (_, msg) = Message::from_wire(&self.ctx, &buf) .map_err(|e| eyre!("Failed to parse BGP message: {e:?}"))?; @@ -90,7 +95,11 @@ impl TestPeer { router_id: Ipv4Addr, hold_time: u16, ) -> Result<()> { - let asn_field = if local_asn > u16::MAX as u32 { 23456 } else { local_asn as u16 }; + let asn_field = if local_asn > u16::MAX as u32 { + 23456 + } else { + local_asn as u16 + }; let open = Message::Open(OpenMessage { version: 4, asn: asn_field, diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index c72d59a..06eca38 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -62,7 +62,14 @@ impl TestServer { let registry = bound.registry.clone(); let handle = tokio::spawn(bound.run()); - Ok(Self { handle, addr, adj_rib_in, local_rib, registry, local_asn }) + Ok(Self { + handle, + addr, + adj_rib_in, + local_rib, + registry, + local_asn, + }) } } diff --git a/crates/integration-tests/tests/announce.rs b/crates/integration-tests/tests/announce.rs index 947de44..5325324 100644 --- a/crates/integration-tests/tests/announce.rs +++ b/crates/integration-tests/tests/announce.rs @@ -67,7 +67,9 @@ async fn test_registry_tracks_session_lifecycle() -> Result<()> { let peer = established_peer(&server).await?; let up = tokio::time::timeout(TIMEOUT, events.recv()).await??; - let SessionEvent::Up(handle) = up else { panic!("expected Up, got {up:?}") }; + let SessionEvent::Up(handle) = up else { + panic!("expected Up, got {up:?}") + }; assert_eq!(handle.peer.asn, PEER_ASN); assert_eq!(handle.local_addr, server.addr); assert!(handle.four_octet_asn); @@ -76,7 +78,10 @@ async fn test_registry_tracks_session_lifecycle() -> Result<()> { drop(peer); let down = tokio::time::timeout(TIMEOUT, events.recv()).await??; - assert!(matches!(down, SessionEvent::Down(p) if p == handle.peer), "got {down:?}"); + assert!( + matches!(down, SessionEvent::Down(p) if p == handle.peer), + "got {down:?}" + ); assert!(server.registry.list().is_empty()); Ok(()) } @@ -102,7 +107,11 @@ async fn test_announce_after_session_up() -> Result<()> { prepend: 0, med: None, communities: vec![(65000, 100), (65535, 65281)], - large_communities: vec![LargeCommunity { global_admin: 65000, data_1: 7, data_2: 8 }], + large_communities: vec![LargeCommunity { + global_admin: 65000, + data_1: 7, + data_2: 8, + }], }, )) .await?; @@ -116,9 +125,16 @@ async fn test_announce_after_session_up() -> Result<()> { PathAttribute::Origin(OriginPathAttribute::IGP), AsPathAttribute::from_asns(vec![LOCAL_ASN]), PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::LOCALHOST)), - PathAttribute::Communities(CommunitiesPathAttribute(vec![(65000, 100), (65535, 65281)])), + PathAttribute::Communities(CommunitiesPathAttribute(vec![ + (65000, 100), + (65535, 65281) + ])), PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute { - communities: vec![LargeCommunity { global_admin: 65000, data_1: 7, data_2: 8 }], + communities: vec![LargeCommunity { + global_admin: 65000, + data_1: 7, + data_2: 8 + }], }), ] ); @@ -130,8 +146,14 @@ async fn test_announce_after_session_up() -> Result<()> { #[tokio::test] async fn test_routes_replayed_on_session_up() -> Result<()> { let server = TestServer::start(LOCAL_ASN).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?; + 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 mut peer = established_peer(&server).await?; @@ -153,7 +175,10 @@ async fn test_announce_reaches_all_sessions() -> Result<()> { let mut b = established_peer_with_id(&server, Ipv4Addr::new(10, 0, 0, 2)).await?; wait_for_sessions(&server, 2).await?; - server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?; + server + .local_rib + .announce(route("10.100.1.0/24", LocalRouteAttrs::default())) + .await?; for peer in [&mut a, &mut b] { let update = next_update(peer).await?; @@ -168,7 +193,10 @@ async fn test_withdraw_sends_withdrawn_route() -> Result<()> { 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.1.0/24", LocalRouteAttrs::default())) + .await?; let _ = next_update(&mut peer).await?; server.local_rib.withdraw(prefix("10.100.1.0/24")).await?; @@ -186,24 +214,39 @@ async fn test_reannounce_with_prepend_and_med() -> Result<()> { 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.1.0/24", LocalRouteAttrs::default())) + .await?; let first = next_update(&mut peer).await?; - assert_eq!(first.path_attributes[1], AsPathAttribute::from_asns(vec![LOCAL_ASN])); + assert_eq!( + first.path_attributes[1], + AsPathAttribute::from_asns(vec![LOCAL_ASN]) + ); server .local_rib .announce(route( "10.100.1.0/24", - LocalRouteAttrs { prepend: 3, med: Some(42), ..LocalRouteAttrs::default() }, + LocalRouteAttrs { + prepend: 3, + med: Some(42), + ..LocalRouteAttrs::default() + }, )) .await?; let second = next_update(&mut peer).await?; assert_eq!(second.announced_routes, vec![prefix("10.100.1.0/24")]); - assert_eq!(second.path_attributes[1], AsPathAttribute::from_asns(vec![LOCAL_ASN; 4])); - assert!(second - .path_attributes - .iter() - .any(|a| matches!(a, PathAttribute::MultiExitDisc(m) if m.0 == 42))); + assert_eq!( + second.path_attributes[1], + AsPathAttribute::from_asns(vec![LOCAL_ASN; 4]) + ); + assert!( + second + .path_attributes + .iter() + .any(|a| matches!(a, PathAttribute::MultiExitDisc(m) if m.0 == 42)) + ); Ok(()) } @@ -212,10 +255,14 @@ async fn test_reannounce_with_prepend_and_med() -> Result<()> { async fn test_keepalives_continue_alongside_updates() -> Result<()> { let server = TestServer::start_with_hold_time(LOCAL_ASN, 3).await?; let mut peer = TestPeer::connect(server.addr).await?; - peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 3).await?; + peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 3) + .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.1.0/24", LocalRouteAttrs::default())) + .await?; let mut saw_update = false; let mut saw_keepalive = false; @@ -243,15 +290,17 @@ async fn test_received_routes_removed_when_session_closes() -> Result<()> { let mut peer = established_peer(&server).await?; wait_for_sessions(&server, 1).await?; - peer.send(&bgp_packet::message::Message::Update(bgp_packet::message::UpdateMessage { - withdrawn_routes: vec![], - path_attributes: vec![ - PathAttribute::Origin(OriginPathAttribute::IGP), - AsPathAttribute::from_asns(vec![PEER_ASN]), - PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(10, 0, 0, 1))), - ], - announced_routes: vec![prefix("203.0.113.0/24"), prefix("198.51.100.0/24")], - })) + peer.send(&bgp_packet::message::Message::Update( + bgp_packet::message::UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + AsPathAttribute::from_asns(vec![PEER_ASN]), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(10, 0, 0, 1))), + ], + announced_routes: vec![prefix("203.0.113.0/24"), prefix("198.51.100.0/24")], + }, + )) .await?; for _ in 0..2 { let ev = tokio::time::timeout(TIMEOUT, rib_events.recv()).await??; @@ -300,8 +349,14 @@ async fn test_route_refresh_resends_local_routes() -> Result<()> { 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?; + 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?; @@ -314,7 +369,11 @@ async fn test_route_refresh_resends_local_routes() -> Result<()> { .collect::>(); resent.sort(); assert_eq!(resent, vec!["10.100.1.0/24", "10.100.2.0/24"]); - assert_eq!(server.registry.list().len(), 1, "session must survive a route refresh"); + assert_eq!( + server.registry.list().len(), + 1, + "session must survive a route refresh" + ); Ok(()) } @@ -345,7 +404,10 @@ async fn test_shutdown_sends_cease_to_every_peer() -> Result<()> { } }) .await??; - assert!(matches!(notif.error, BgpError::Cease(CeaseSubcode::AdminShutdown)), "{notif:?}"); + assert!( + matches!(notif.error, BgpError::Cease(CeaseSubcode::AdminShutdown)), + "{notif:?}" + ); assert!(peer.is_closed().await); } Ok(()) @@ -359,8 +421,15 @@ async fn test_shutdown_sends_cease_to_every_peer() -> Result<()> { /// connects, and each session is sent them when it comes up. #[tokio::test] async fn test_configured_routes_are_announced_to_every_new_session() -> Result<()> { - let attrs = LocalRouteAttrs { med: Some(7), communities: vec![(65000, 100)], ..Default::default() }; - let configured = vec![route("10.100.1.0/24", attrs), route("10.100.2.0/24", LocalRouteAttrs::default())]; + let attrs = LocalRouteAttrs { + med: Some(7), + communities: vec![(65000, 100)], + ..Default::default() + }; + let configured = vec![ + route("10.100.1.0/24", attrs), + 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); @@ -369,7 +438,13 @@ async fn test_configured_routes_are_announced_to_every_new_session() -> Result<( let mut seen = vec![next_update(&mut peer).await?, next_update(&mut peer).await?]; seen.sort_by_key(|u| u.announced_routes[0].to_string()); assert_eq!(seen[0].announced_routes, vec![prefix("10.100.1.0/24")]); - assert!(seen[0].path_attributes.contains(&PathAttribute::Communities(CommunitiesPathAttribute(vec![(65000, 100)])))); + assert!( + seen[0] + .path_attributes + .contains(&PathAttribute::Communities(CommunitiesPathAttribute(vec![ + (65000, 100) + ]))) + ); assert_eq!(seen[1].announced_routes, vec![prefix("10.100.2.0/24")]); } Ok(()) diff --git a/crates/integration-tests/tests/basic_connection.rs b/crates/integration-tests/tests/basic_connection.rs index c69ed82..b3bd139 100644 --- a/crates/integration-tests/tests/basic_connection.rs +++ b/crates/integration-tests/tests/basic_connection.rs @@ -222,7 +222,8 @@ async fn test_rib_records_ipv4_announce() -> Result<()> { let mut events = server.adj_rib_in.subscribe(); let mut peer = TestPeer::connect(server.addr).await?; - peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?; peer.send(&Message::Update(UpdateMessage { @@ -253,7 +254,8 @@ async fn test_rib_records_ipv4_withdraw() -> Result<()> { let mut events = server.adj_rib_in.subscribe(); let mut peer = TestPeer::connect(server.addr).await?; - peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?; @@ -295,7 +297,8 @@ async fn test_rib_records_ipv6_announce() -> Result<()> { let mut events = server.adj_rib_in.subscribe(); let mut peer = TestPeer::connect(server.addr).await?; - peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?; peer.send(&Message::Update(UpdateMessage { @@ -331,7 +334,8 @@ async fn test_rib_records_ipv6_withdraw() -> Result<()> { let mut events = server.adj_rib_in.subscribe(); let mut peer = TestPeer::connect(server.addr).await?; - peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?; diff --git a/crates/integration-tests/tests/collision.rs b/crates/integration-tests/tests/collision.rs index b451b18..5e8165b 100644 --- a/crates/integration-tests/tests/collision.rs +++ b/crates/integration-tests/tests/collision.rs @@ -33,7 +33,13 @@ async fn server_dialing(listener: &TcpListener) -> Result { let endpoint: SocketAddr = listener.local_addr()?; TestServer::start_with(LOCAL_ASN, |c| { c.connect_retry = Duration::from_millis(200); - c.peers = vec![PeerConfig { name: "r1".into(), endpoint, remote_asn: PEER_ASN, hold_time_secs: 90, local_address: None }]; + c.peers = vec![PeerConfig { + name: "r1".into(), + endpoint, + remote_asn: PEER_ASN, + hold_time_secs: 90, + local_address: None, + }]; }) .await } @@ -70,7 +76,10 @@ async fn expect_notification(peer: &mut TestPeer) -> Result } fn is_collision_cease(n: &NotificationMessage) -> bool { - matches!(n.error, BgpError::Cease(CeaseSubcode::ConnCollisionResolution)) + matches!( + n.error, + BgpError::Cease(CeaseSubcode::ConnCollisionResolution) + ) } async fn wait_for_sessions(server: &TestServer, n: usize) -> Result<()> { @@ -112,14 +121,20 @@ async fn test_simultaneous_open_peer_higher_keeps_inbound() -> Result<()> { // The survivor is fully functional. server .local_rib - .announce(LocalRoute::new(IpPrefix::try_from("10.100.1.0/24")?, LocalRouteAttrs::default())) + .announce(LocalRoute::new( + IpPrefix::try_from("10.100.1.0/24")?, + LocalRouteAttrs::default(), + )) .await?; let update = tokio::time::timeout(TIMEOUT, inbound.recv_update()).await??; assert_eq!(update.announced_routes[0].to_string(), "10.100.1.0/24"); // The dialer notices the surviving session and does not open another. let extra = tokio::time::timeout(Duration::from_millis(700), listener.accept()).await; - assert!(extra.is_err(), "dialer opened a second connection despite an established session"); + assert!( + extra.is_err(), + "dialer opened a second connection despite an established session" + ); Ok(()) } @@ -190,7 +205,10 @@ async fn test_established_session_wins_by_default() -> Result<()> { assert_eq!(server.registry.list().len(), 1); server .local_rib - .announce(LocalRoute::new(IpPrefix::try_from("10.100.1.0/24")?, LocalRouteAttrs::default())) + .announce(LocalRoute::new( + IpPrefix::try_from("10.100.1.0/24")?, + LocalRouteAttrs::default(), + )) .await?; let update = tokio::time::timeout(TIMEOUT, first.recv_update()).await??; assert_eq!(update.announced_routes[0].to_string(), "10.100.1.0/24"); @@ -202,7 +220,8 @@ async fn test_established_session_wins_by_default() -> Result<()> { /// old session are flushed. #[tokio::test] async fn test_restarted_peer_replaces_stale_session_when_enabled() -> Result<()> { - let server = TestServer::start_with(LOCAL_ASN, |c| c.collision_detect_established_state = true).await?; + let server = + TestServer::start_with(LOCAL_ASN, |c| c.collision_detect_established_state = true).await?; let mut stale = TestPeer::connect(server.addr).await?; stale.handshake(PEER_ASN, HIGH_ID, 90).await?; @@ -243,7 +262,10 @@ async fn test_restarted_peer_replaces_stale_session_when_enabled() -> Result<()> } }) .await?; - assert!(server.adj_rib_in.snapshot().await.is_empty(), "stale routes were not flushed"); + assert!( + server.adj_rib_in.snapshot().await.is_empty(), + "stale routes were not flushed" + ); Ok(()) } @@ -282,6 +304,9 @@ async fn test_peer_with_our_identifier_is_refused() -> Result<()> { let mut peer = TestPeer::connect(server.addr).await?; open_only(&mut peer, Ipv4Addr::new(127, 0, 0, 1)).await?; let notif = expect_notification(&mut peer).await?; - assert!(matches!(notif.error, BgpError::OpenMsg(OpenMsgSubcode::BadBgpId)), "{notif:?}"); + assert!( + matches!(notif.error, BgpError::OpenMsg(OpenMsgSubcode::BadBgpId)), + "{notif:?}" + ); Ok(()) } diff --git a/crates/integration-tests/tests/dialer.rs b/crates/integration-tests/tests/dialer.rs index b45d3ce..6d8f1d2 100644 --- a/crates/integration-tests/tests/dialer.rs +++ b/crates/integration-tests/tests/dialer.rs @@ -24,7 +24,13 @@ async fn server_dialing(listener: &TcpListener, remote_asn: u32) -> Result Result<()> { let mut peer = accept(&listener).await?; // Both sides send OPEN on connect; the test peer's handshake order works // whichever side opened TCP. - let open = peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + let open = peer + .handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90) + .await?; assert_eq!(open.asn, LOCAL_ASN as u16); let mut events = server.registry.subscribe(); @@ -61,14 +69,25 @@ async fn test_server_dials_configured_peer_and_establishes() -> Result<()> { assert_eq!(listed.peer.addr, listener.local_addr()?); // Configured view reflects it; the dialer is not in backoff. - let (cfg, state) = server.registry.configured_peers().into_iter().next().unwrap(); + let (cfg, state) = server + .registry + .configured_peers() + .into_iter() + .next() + .unwrap(); assert_eq!(cfg.name, "r1"); - assert!(matches!(state, DialState::Connecting | DialState::Idle), "{state:?}"); + assert!( + matches!(state, DialState::Connecting | DialState::Idle), + "{state:?}" + ); // Local routes flow over the dialled session too. server .local_rib - .announce(LocalRoute::new(IpPrefix::try_from("10.100.1.0/24")?, LocalRouteAttrs::default())) + .announce(LocalRoute::new( + IpPrefix::try_from("10.100.1.0/24")?, + LocalRouteAttrs::default(), + )) .await?; let update = tokio::time::timeout(TIMEOUT, peer.recv_update()).await??; assert_eq!(update.announced_routes[0].to_string(), "10.100.1.0/24"); @@ -81,7 +100,9 @@ async fn test_server_reconnects_after_peer_closes() -> Result<()> { let server = server_dialing(&listener, PEER_ASN).await?; let mut first = accept(&listener).await?; - first.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + first + .handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90) + .await?; tokio::time::timeout(TIMEOUT, async { while server.registry.list().is_empty() { tokio::time::sleep(Duration::from_millis(10)).await; @@ -97,11 +118,18 @@ async fn test_server_reconnects_after_peer_closes() -> Result<()> { } }) .await?; - let (_, state) = server.registry.configured_peers().into_iter().next().unwrap(); + let (_, state) = server + .registry + .configured_peers() + .into_iter() + .next() + .unwrap(); assert!(matches!(state, DialState::Backoff { .. }), "{state:?}"); let mut second = accept(&listener).await?; - second.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + second + .handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90) + .await?; tokio::time::timeout(TIMEOUT, async { while server.registry.list().is_empty() { tokio::time::sleep(Duration::from_millis(10)).await; @@ -117,7 +145,8 @@ async fn test_wrong_asn_is_refused_with_bad_peer_as() -> Result<()> { let _server = server_dialing(&listener, PEER_ASN).await?; let mut peer = accept(&listener).await?; - peer.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + peer.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90) + .await?; let _ = peer.recv_open().await?; let notif = tokio::time::timeout(TIMEOUT, async { loop { @@ -129,7 +158,10 @@ async fn test_wrong_asn_is_refused_with_bad_peer_as() -> Result<()> { } }) .await??; - assert!(matches!(notif.error, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs)), "{notif:?}"); + assert!( + matches!(notif.error, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs)), + "{notif:?}" + ); // The dialer retries rather than giving up. let _again = accept(&listener).await?; @@ -159,11 +191,15 @@ async fn test_inbound_from_configured_address_uses_config() -> Result<()> { bad.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90).await?; let _ = bad.recv_open().await?; let msg = tokio::time::timeout(TIMEOUT, bad.recv()).await??; - assert!(matches!(&msg, Message::Notification(n) if matches!(n.error, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs))), "{msg:?}"); + assert!( + matches!(&msg, Message::Notification(n) if matches!(n.error, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs))), + "{msg:?}" + ); // Right ASN inbound → established, named. let mut good = TestPeer::connect(server.addr).await?; - good.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + good.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90) + .await?; let listed = tokio::time::timeout(TIMEOUT, async { loop { if let Some(s) = server.registry.list().first().cloned() { @@ -212,7 +248,9 @@ async fn test_local_address_is_used_as_the_source() -> Result<()> { .await?; let reason = tokio::time::timeout(TIMEOUT, async { loop { - if let Some((_, DialState::Backoff { last_error, .. })) = server.registry.configured_peers().into_iter().next() { + if let Some((_, DialState::Backoff { last_error, .. })) = + server.registry.configured_peers().into_iter().next() + { return last_error; } tokio::time::sleep(Duration::from_millis(20)).await; diff --git a/crates/integration-tests/tests/http_api.rs b/crates/integration-tests/tests/http_api.rs index 8953e5e..81513be 100644 --- a/crates/integration-tests/tests/http_api.rs +++ b/crates/integration-tests/tests/http_api.rs @@ -24,7 +24,8 @@ fn prefix(s: &str) -> IpPrefix { async fn established_peer(server: &TestServer) -> Result { let mut peer = TestPeer::connect(server.addr).await?; - peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 90).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; @@ -35,7 +36,13 @@ async fn established_peer(server: &TestServer) -> Result { } async fn get_json(client: &reqwest::Client, url: &str) -> Result { - Ok(client.get(url).send().await?.error_for_status()?.json().await?) + Ok(client + .get(url) + .send() + .await? + .error_for_status()? + .json() + .await?) } #[tokio::test] @@ -75,7 +82,11 @@ async fn test_originated_routes_are_listed() -> Result<()> { prepend: 1, med: Some(10), communities: vec![(65000, 100), (65535, 65281)], - large_communities: vec![LargeCommunity { global_admin: 65000, data_1: 1, data_2: 2 }], + large_communities: vec![LargeCommunity { + global_admin: 65000, + data_1: 1, + data_2: 2, + }], ..Default::default() }; let server = TestServer::start_with(LOCAL_ASN, |c| { @@ -93,7 +104,10 @@ async fn test_originated_routes_are_listed() -> Result<()> { assert_eq!(local[0]["med"], 10); assert_eq!(local[0]["communities"], json!(["65000:100", "65535:65281"])); assert_eq!(local[0]["large_communities"], json!(["65000:1:2"])); - assert_eq!(get_json(&client, &format!("{base}/api/status")).await?["local_routes"], 1); + assert_eq!( + get_json(&client, &format!("{base}/api/status")).await?["local_routes"], + 1 + ); Ok(()) } @@ -106,10 +120,23 @@ async fn test_api_is_read_only() -> Result<()> { let client = reqwest::Client::new(); let body = json!({ "prefix": "10.100.1.0/24" }); - for path in ["/api/routes/local", "/api/routes/received", "/api/peers", "/api/status"] { - let post = client.post(format!("{base}{path}")).json(&body).send().await?; + for path in [ + "/api/routes/local", + "/api/routes/received", + "/api/peers", + "/api/status", + ] { + let post = client + .post(format!("{base}{path}")) + .json(&body) + .send() + .await?; assert_eq!(post.status(), 405, "POST {path}"); - let delete = client.delete(format!("{base}{path}")).json(&body).send().await?; + let delete = client + .delete(format!("{base}{path}")) + .json(&body) + .send() + .await?; assert_eq!(delete.status(), 405, "DELETE {path}"); } assert!(server.local_rib.snapshot().await.is_empty()); @@ -161,8 +188,16 @@ async fn test_event_stream() -> Result<()> { let (base, _http) = server.serve_http().await?; let client = reqwest::Client::new(); - let mut resp = client.get(format!("{base}/api/events")).send().await?.error_for_status()?; - assert!(resp.headers()["content-type"].to_str()?.starts_with("text/event-stream")); + let mut resp = client + .get(format!("{base}/api/events")) + .send() + .await? + .error_for_status()?; + assert!( + resp.headers()["content-type"] + .to_str()? + .starts_with("text/event-stream") + ); let _peer = established_peer(&server).await?; server @@ -176,13 +211,18 @@ async fn test_event_stream() -> Result<()> { let mut seen_announce = false; tokio::time::timeout(TIMEOUT, async { while !(seen_up && seen_announce) { - let chunk = resp.chunk().await?.ok_or_else(|| eyre::eyre!("stream ended"))?; + let chunk = resp + .chunk() + .await? + .ok_or_else(|| eyre::eyre!("stream ended"))?; buf.push_str(std::str::from_utf8(&chunk)?); for line in buf.lines().filter_map(|l| l.strip_prefix("data:")) { let ev: Value = serde_json::from_str(line.trim())?; match ev["type"].as_str() { Some("session_up") if ev["peer"]["peer_asn"] == PEER_ASN => seen_up = true, - Some("local_announced") if ev["route"]["prefix"] == "10.100.1.0/24" => seen_announce = true, + Some("local_announced") if ev["route"]["prefix"] == "10.100.1.0/24" => { + seen_announce = true + } _ => {} } } @@ -201,13 +241,19 @@ async fn test_statusz_page_is_served() -> Result<()> { let resp = client.get(format!("{base}/statusz")).send().await?; assert_eq!(resp.status(), 200); - assert!(resp.headers()["content-type"].to_str()?.starts_with("text/html")); + assert!( + resp.headers()["content-type"] + .to_str()? + .starts_with("text/html") + ); let body = resp.text().await?; assert!(body.contains("Status for ferrix_bgp")); assert!(body.contains("/api/events")); // The root redirects to the page. - let no_redirect = reqwest::Client::builder().redirect(reqwest::redirect::Policy::none()).build()?; + let no_redirect = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build()?; let resp = no_redirect.get(format!("{base}/")).send().await?; assert_eq!(resp.status(), 307); assert_eq!(resp.headers()["location"], "/statusz"); @@ -221,7 +267,13 @@ async fn test_metricz() -> Result<()> { let client = reqwest::Client::new(); let _peer = established_peer(&server).await?; - let text = client.get(format!("{base}/metricz")).send().await?.error_for_status()?.text().await?; + let text = client + .get(format!("{base}/metricz")) + .send() + .await? + .error_for_status()? + .text() + .await?; // Metrics are process-global, so other tests in this binary contribute // too; assert on presence and a lower bound rather than exact values. let metric = |name: &str| -> Option { @@ -229,9 +281,18 @@ async fn test_metricz() -> Result<()> { .find(|l| l.starts_with(name) && l[name.len()..].starts_with(' ')) .and_then(|l| l.rsplit(' ').next()?.parse().ok()) }; - assert!(metric("bgp_sessions_established").unwrap_or(0) >= 1, "{text}"); - assert!(metric("bgp_messages_sent_total{type=\"open\"}").unwrap_or(0) >= 1, "{text}"); - assert!(metric("bgp_messages_received_total{type=\"keepalive\"}").unwrap_or(0) >= 1, "{text}"); + assert!( + metric("bgp_sessions_established").unwrap_or(0) >= 1, + "{text}" + ); + assert!( + metric("bgp_messages_sent_total{type=\"open\"}").unwrap_or(0) >= 1, + "{text}" + ); + assert!( + metric("bgp_messages_received_total{type=\"keepalive\"}").unwrap_or(0) >= 1, + "{text}" + ); assert!(metric("bgp_local_routes").is_some(), "{text}"); assert!(metric("bgp_adj_rib_in_routes").is_some(), "{text}"); Ok(()) diff --git a/crates/integration-tests/tests/interop.rs b/crates/integration-tests/tests/interop.rs index b6c3d59..3b252b3 100644 --- a/crates/integration-tests/tests/interop.rs +++ b/crates/integration-tests/tests/interop.rs @@ -90,14 +90,21 @@ async fn test_bird_open_and_end_of_rib_are_accepted() -> Result<()> { }) .await?; let session = server.registry.list().remove(0); - assert_eq!(session.peer.asn, 210036, "4-octet ASN must come from the capability, not AS_TRANS"); + assert_eq!( + session.peer.asn, 210036, + "4-octet ASN must come from the capability, not AS_TRANS" + ); assert!(session.four_octet_asn); // End-of-RIB must neither break the session nor create a route. peer.send_raw(IPV6_END_OF_RIB).await?; peer.send(&Message::KeepAlive).await?; tokio::time::sleep(Duration::from_millis(200)).await; - assert_eq!(server.registry.list().len(), 1, "session dropped after End-of-RIB"); + assert_eq!( + server.registry.list().len(), + 1, + "session dropped after End-of-RIB" + ); assert!(server.adj_rib_in.snapshot().await.is_empty()); Ok(()) } @@ -107,13 +114,27 @@ async fn test_bird_open_and_end_of_rib_are_accepted() -> Result<()> { /// an IPv6 prefix carries MP_REACH_NLRI with an IPv6 next hop and no NEXT_HOP. #[tokio::test] async fn test_ipv6_prefix_update_shape() -> Result<()> { - let route = LocalRoute::new(IpPrefix::try_from("2001:db8:1::/48")?, LocalRouteAttrs::default()); + let route = LocalRoute::new( + IpPrefix::try_from("2001:db8:1::/48")?, + LocalRouteAttrs::default(), + ); let nh: std::net::Ipv6Addr = "fd00:179::2".parse()?; - let update = bgp_server::advertise::build_announce(&route, 4200000001, nh.into()).expect("ipv6 over ipv6"); + let update = bgp_server::advertise::build_announce(&route, 4200000001, nh.into()) + .expect("ipv6 over ipv6"); assert!(update.announced_routes.is_empty()); - assert!(!update.path_attributes.iter().any(|a| matches!(a, PathAttribute::NextHop(_)))); - assert!(update.path_attributes.iter().any(|a| matches!(a, PathAttribute::MpReachNlri(m) - if m.afi == AddressFamilyId::Ipv6 && m.prefixes[0].to_string() == "2001:db8:1::/48"))); + assert!( + !update + .path_attributes + .iter() + .any(|a| matches!(a, PathAttribute::NextHop(_))) + ); + assert!( + update + .path_attributes + .iter() + .any(|a| matches!(a, PathAttribute::MpReachNlri(m) + if m.afi == AddressFamilyId::Ipv6 && m.prefixes[0].to_string() == "2001:db8:1::/48")) + ); let _ = Ipv4Addr::LOCALHOST; Ok(()) } diff --git a/crates/packet/src/community.rs b/crates/packet/src/community.rs index a94f486..d078feb 100644 --- a/crates/packet/src/community.rs +++ b/crates/packet/src/community.rs @@ -19,7 +19,10 @@ pub const WELL_KNOWN: &[(&str, (u16, u16))] = &[ pub fn parse_standard(s: &str) -> Result<(u16, u16), String> { let s = s.trim(); - if let Some((_, value)) = WELL_KNOWN.iter().find(|(name, _)| name.eq_ignore_ascii_case(s)) { + if let Some((_, value)) = WELL_KNOWN + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case(s)) + { return Ok(*value); } let (high, low) = s @@ -41,14 +44,20 @@ pub fn parse_large(s: &str) -> Result { let s = s.trim(); let parts: Vec<&str> = s.split(':').collect(); if parts.len() != 3 { - return Err(format!("large community {s:?} must be GLOBAL:LOCAL1:LOCAL2")); + return Err(format!( + "large community {s:?} must be GLOBAL:LOCAL1:LOCAL2" + )); } let parse = |part: &str| { part.trim() .parse::() .map_err(|_| format!("large community {s:?}: {part:?} is not a 32-bit number")) }; - Ok(LargeCommunity { global_admin: parse(parts[0])?, data_1: parse(parts[1])?, data_2: parse(parts[2])? }) + Ok(LargeCommunity { + global_admin: parse(parts[0])?, + data_1: parse(parts[1])?, + data_2: parse(parts[2])?, + }) } pub fn format_large(c: &LargeCommunity) -> String { @@ -73,7 +82,14 @@ mod tests { #[test] fn large_roundtrip() { let c = parse_large("65000:1:2").unwrap(); - assert_eq!(c, LargeCommunity { global_admin: 65000, data_1: 1, data_2: 2 }); + assert_eq!( + c, + LargeCommunity { + global_admin: 65000, + data_1: 1, + data_2: 2 + } + ); assert_eq!(format_large(&c), "65000:1:2"); assert!(parse_large("65000:1").is_err()); assert!(parse_large("a:b:c").is_err()); diff --git a/crates/packet/src/open.rs b/crates/packet/src/open.rs index 827624e..3a77763 100644 --- a/crates/packet/src/open.rs +++ b/crates/packet/src/open.rs @@ -993,11 +993,17 @@ mod tests { let samples: &[(&str, &[u8])] = &[ ("multiprotocol", &[0x01, 0x04, 0x00, 0x02, 0x00, 0x01]), ("route refresh", &[0x02, 0x00]), - ("extended next hop", &[0x05, 0x06, 0x00, 0x01, 0x00, 0x01, 0x00, 0x02]), + ( + "extended next hop", + &[0x05, 0x06, 0x00, 0x01, 0x00, 0x01, 0x00, 0x02], + ), ("extended message", &[0x06, 0x00]), ("role", &[0x09, 0x01, 0x00]), ("graceful restart, aware only", &[0x40, 0x02, 0x00, 0x00]), - ("graceful restart, one family", &[0x40, 0x06, 0x00, 0x78, 0x00, 0x02, 0x01, 0x80]), + ( + "graceful restart, one family", + &[0x40, 0x06, 0x00, 0x78, 0x00, 0x02, 0x01, 0x80], + ), ("4-octet asn", &[0x41, 0x04, 0x00, 0x00, 0xfd, 0xe8]), ("add-path", &[0x45, 0x04, 0x00, 0x02, 0x01, 0x03]), ("enhanced route refresh", &[0x46, 0x00]), @@ -1009,7 +1015,10 @@ mod tests { input.extend_from_slice(trailer); let (rest, _) = Capability::from_wire(&ctx, &input) .unwrap_or_else(|e| panic!("{name}: failed to parse: {e:?}")); - assert_eq!(rest, trailer, "{name}: did not return the bytes following the capability"); + assert_eq!( + rest, trailer, + "{name}: did not return the bytes following the capability" + ); } } @@ -1033,5 +1042,4 @@ mod tests { assert_eq!(caps.len(), 7, "{caps:?}"); assert!(caps.contains(&Capability::FourByteAsn { asn: 210036 })); } - } diff --git a/crates/rib/src/adj_rib_in.rs b/crates/rib/src/adj_rib_in.rs index ea9e56c..20bb0bb 100644 --- a/crates/rib/src/adj_rib_in.rs +++ b/crates/rib/src/adj_rib_in.rs @@ -11,11 +11,23 @@ const EVENT_CHANNEL_CAPACITY: usize = 1024; const CMD_CHANNEL_CAPACITY: usize = 256; enum RibCommand { - Announce { peer: PeerId, prefix: IpPrefix, entry: RibEntry }, - Withdraw { peer: PeerId, prefix: IpPrefix }, + Announce { + peer: PeerId, + prefix: IpPrefix, + entry: RibEntry, + }, + Withdraw { + peer: PeerId, + prefix: IpPrefix, + }, /// Drop everything learned from `peer`, e.g. because its session ended. - RemovePeer { peer: PeerId, reply: oneshot::Sender }, - Snapshot { reply: oneshot::Sender> }, + RemovePeer { + peer: PeerId, + reply: oneshot::Sender, + }, + Snapshot { + reply: oneshot::Sender>, + }, } /// Handle to the Adj-RIB-In actor. Cheap to clone — all clones share the same @@ -37,7 +49,14 @@ impl AdjRibIn { } pub async fn announce(&self, peer: PeerId, prefix: IpPrefix, entry: RibEntry) { - let _ = self.tx.send(RibCommand::Announce { peer, prefix, entry }).await; + let _ = self + .tx + .send(RibCommand::Announce { + peer, + prefix, + entry, + }) + .await; } pub async fn withdraw(&self, peer: PeerId, prefix: IpPrefix) { @@ -49,7 +68,13 @@ impl AdjRibIn { /// 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; + let _ = self + .tx + .send(RibCommand::RemovePeer { + peer, + reply: reply_tx, + }) + .await; reply_rx.await.unwrap_or(0) } @@ -76,9 +101,17 @@ async fn run_actor(mut rx: mpsc::Receiver, events: broadcast::Sender while let Some(cmd) = rx.recv().await { match cmd { // Send errors below mean there are no subscribers, which is fine. - RibCommand::Announce { peer, prefix, entry } => { + RibCommand::Announce { + peer, + prefix, + entry, + } => { table.insert((peer, prefix.clone()), entry.clone()); - let _ = events.send(RibEvent::Announced { peer, prefix, entry }); + let _ = events.send(RibEvent::Announced { + peer, + prefix, + entry, + }); } RibCommand::Withdraw { peer, prefix } => { if table.remove(&(peer, prefix.clone())).is_some() { @@ -86,11 +119,17 @@ async fn run_actor(mut rx: mpsc::Receiver, events: broadcast::Sender } } RibCommand::RemovePeer { peer, reply } => { - let prefixes: Vec = - table.keys().filter(|(p, _)| *p == peer).map(|(_, prefix)| prefix.clone()).collect(); + let prefixes: Vec = table + .keys() + .filter(|(p, _)| *p == peer) + .map(|(_, prefix)| prefix.clone()) + .collect(); for prefix in &prefixes { table.remove(&(peer, prefix.clone())); - let _ = events.send(RibEvent::Withdrawn { peer, prefix: prefix.clone() }); + let _ = events.send(RibEvent::Withdrawn { + peer, + prefix: prefix.clone(), + }); } let _ = reply.send(prefixes.len()); } diff --git a/crates/rib/src/entry.rs b/crates/rib/src/entry.rs index bdf278e..43d051c 100644 --- a/crates/rib/src/entry.rs +++ b/crates/rib/src/entry.rs @@ -26,7 +26,10 @@ pub struct RibEntry { impl RibEntry { pub fn new(path_attributes: Vec) -> Self { - Self { path_attributes, received_at: SystemTime::now() } + Self { + path_attributes, + received_at: SystemTime::now(), + } } } diff --git a/crates/rib/src/event.rs b/crates/rib/src/event.rs index 868afdd..3716b25 100644 --- a/crates/rib/src/event.rs +++ b/crates/rib/src/event.rs @@ -4,6 +4,13 @@ use crate::entry::{PeerId, RibEntry}; #[derive(Debug, Clone)] pub enum RibEvent { - Announced { peer: PeerId, prefix: IpPrefix, entry: RibEntry }, - Withdrawn { peer: PeerId, prefix: IpPrefix }, + Announced { + peer: PeerId, + prefix: IpPrefix, + entry: RibEntry, + }, + Withdrawn { + peer: PeerId, + prefix: IpPrefix, + }, } diff --git a/crates/rib/src/local_rib.rs b/crates/rib/src/local_rib.rs index 58140e7..e003ed2 100644 --- a/crates/rib/src/local_rib.rs +++ b/crates/rib/src/local_rib.rs @@ -53,7 +53,11 @@ pub struct LocalRoute { impl LocalRoute { pub fn new(prefix: IpPrefix, attrs: LocalRouteAttrs) -> Self { - Self { prefix, attrs, announced_at: SystemTime::now() } + Self { + prefix, + attrs, + announced_at: SystemTime::now(), + } } } @@ -67,13 +71,23 @@ pub enum LocalRibError { pub enum LocalRibEvent { /// A route was added or its attributes replaced. Announced(LocalRoute), - Withdrawn { prefix: IpPrefix }, + Withdrawn { + prefix: IpPrefix, + }, } enum Command { - Announce { route: LocalRoute, reply: oneshot::Sender<()> }, - Withdraw { prefix: IpPrefix, reply: oneshot::Sender> }, - Snapshot { reply: oneshot::Sender> }, + Announce { + route: LocalRoute, + reply: oneshot::Sender<()>, + }, + Withdraw { + prefix: IpPrefix, + reply: oneshot::Sender>, + }, + Snapshot { + reply: oneshot::Sender>, + }, } /// Handle to the local RIB actor. Cheap to clone; all clones share one table. @@ -170,12 +184,16 @@ mod tests { let mut events = rib.subscribe(); rib.announce(route("10.0.0.0/24")).await.unwrap(); - assert!(matches!(events.recv().await.unwrap(), LocalRibEvent::Announced(r) if r.prefix == prefix("10.0.0.0/24"))); + 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); 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!( + matches!(events.recv().await.unwrap(), LocalRibEvent::Withdrawn { prefix: p } if p == prefix("10.0.0.0/24")) + ); assert!(rib.snapshot().await.is_empty()); // Withdrawing an absent prefix is a no-op, not an error. @@ -186,8 +204,13 @@ mod tests { async fn announcing_a_prefix_again_replaces_its_attributes() { let rib = LocalRib::spawn(); rib.announce(route("10.0.0.0/24")).await.unwrap(); - let attrs = LocalRouteAttrs { med: Some(50), ..Default::default() }; - rib.announce(LocalRoute::new(prefix("10.0.0.0/24"), attrs.clone())).await.unwrap(); + let attrs = LocalRouteAttrs { + med: Some(50), + ..Default::default() + }; + rib.announce(LocalRoute::new(prefix("10.0.0.0/24"), attrs.clone())) + .await + .unwrap(); let table = rib.snapshot().await; assert_eq!(table.len(), 1); diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 94a3b27..bbd1bda 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -117,7 +117,11 @@ async fn send_withdraw(session: &SessionHandle, prefix: &IpPrefix) { /// Returns `None` when the prefix family does not match the next-hop family: /// carrying IPv4 NLRI over an IPv6 next hop (RFC 8950) or vice versa is not /// supported yet. -pub fn build_announce(route: &LocalRoute, local_asn: u32, next_hop: IpAddr) -> Option { +pub fn build_announce( + route: &LocalRoute, + local_asn: u32, + next_hop: IpAddr, +) -> Option { let attrs = &route.attrs; // Path attributes in ascending type-code order, as most implementations @@ -138,7 +142,9 @@ pub fn build_announce(route: &LocalRoute, local_asn: u32, next_hop: IpAddr) -> O } if let Some(med) = attrs.med { - path_attributes.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(med))); + path_attributes.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute( + med, + ))); } if !attrs.communities.is_empty() { path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute( @@ -154,12 +160,18 @@ pub fn build_announce(route: &LocalRoute, local_asn: u32, next_hop: IpAddr) -> O })); } if !attrs.large_communities.is_empty() { - path_attributes.push(PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute { - communities: attrs.large_communities.clone(), - })); + path_attributes.push(PathAttribute::LargeCommunities( + LargeCommunitiesPathAttribute { + communities: attrs.large_communities.clone(), + }, + )); } - Some(UpdateMessage { withdrawn_routes: vec![], path_attributes, announced_routes }) + Some(UpdateMessage { + withdrawn_routes: vec![], + path_attributes, + announced_routes, + }) } /// Builds the UPDATE withdrawing `prefix`. @@ -206,10 +218,15 @@ mod tests { prepend: 2, med: Some(50), communities: vec![(65000, 100), (65535, 65281)], - large_communities: vec![LargeCommunity { global_admin: 65000, data_1: 1, data_2: 2 }], + large_communities: vec![LargeCommunity { + global_admin: 65000, + data_1: 1, + data_2: 2, + }], }, ); - let update = build_announce(&route, 65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1))).unwrap(); + let update = + build_announce(&route, 65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1))).unwrap(); assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]); assert!(update.withdrawn_routes.is_empty()); @@ -220,20 +237,32 @@ mod tests { AsPathAttribute::from_asns(vec![65000, 65000, 65000]), PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))), PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(50)), - PathAttribute::Communities(CommunitiesPathAttribute(vec![(65000, 100), (65535, 65281)])), + PathAttribute::Communities(CommunitiesPathAttribute(vec![ + (65000, 100), + (65535, 65281) + ])), PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute { - communities: vec![LargeCommunity { global_admin: 65000, data_1: 1, data_2: 2 }], + communities: vec![LargeCommunity { + global_admin: 65000, + data_1: 1, + data_2: 2 + }], }), ] ); // Serialise and parse back to make sure the wire encoding is sound. - let ctx = ParserContext { four_octet_asn: Some(true), address_family: None }; + let ctx = ParserContext { + four_octet_asn: Some(true), + address_family: None, + }; let msg = Message::Update(update.clone()); let mut buf = BytesMut::new(); msg.to_wire(&ctx, &mut buf).unwrap(); let (_, parsed) = Message::from_wire(&ctx, &buf).unwrap(); - let Message::Update(parsed) = parsed else { panic!("expected UPDATE") }; + let Message::Update(parsed) = parsed else { + panic!("expected UPDATE") + }; assert_eq!(parsed, update); } @@ -244,7 +273,12 @@ mod tests { let update = build_announce(&route, 65000, IpAddr::V6(nh)).unwrap(); assert!(update.announced_routes.is_empty()); - assert!(!update.path_attributes.iter().any(|a| matches!(a, PathAttribute::NextHop(_)))); + assert!( + !update + .path_attributes + .iter() + .any(|a| matches!(a, PathAttribute::NextHop(_))) + ); assert!(update.path_attributes.iter().any(|a| matches!( a, PathAttribute::MpReachNlri(MpReachNlriPathAttribute { next_hop: NlriNextHop::Ipv6(h), prefixes, .. }) @@ -268,6 +302,8 @@ mod tests { let v6 = build_withdraw(&prefix("2001:db8::/32")); assert!(v6.withdrawn_routes.is_empty()); - assert!(matches!(&v6.path_attributes[..], [PathAttribute::MpUnreachNlri(m)] if m.prefixes == vec![prefix("2001:db8::/32")])); + assert!( + matches!(&v6.path_attributes[..], [PathAttribute::MpUnreachNlri(m)] if m.prefixes == vec![prefix("2001:db8::/32")]) + ); } } diff --git a/crates/server/src/collision.rs b/crates/server/src/collision.rs index 54b197b..0226cee 100644 --- a/crates/server/src/collision.rs +++ b/crates/server/src/collision.rs @@ -67,7 +67,11 @@ fn new_loses( return false; } // Identifiers compare as unsigned 32-bit integers. - let keep = if u32::from(local_id) > u32::from(remote_id) { Direction::Outbound } else { Direction::Inbound }; + let keep = if u32::from(local_id) > u32::from(remote_id) { + Direction::Outbound + } else { + Direction::Inbound + }; new != keep } @@ -81,17 +85,32 @@ pub struct CollisionTable { impl CollisionTable { pub fn new(detect_established: bool) -> Self { - Self { entries: Default::default(), next_id: Default::default(), detect_established } + Self { + entries: Default::default(), + next_id: Default::default(), + detect_established, + } } /// Call when a connection receives the peer's OPEN. `None` means this /// connection lost and must be closed with Cease / Connection Collision /// Resolution. Otherwise any rival has been told to close and this /// connection holds the slot until the guard drops. - pub fn admit(&self, local_id: Ipv4Addr, remote_id: Ipv4Addr, direction: Direction) -> Option { + pub fn admit( + &self, + local_id: Ipv4Addr, + remote_id: Ipv4Addr, + direction: Direction, + ) -> Option { let mut entries = self.entries.lock().expect("collision table poisoned"); if let Some(existing) = entries.get_mut(&remote_id) { - if new_loses(local_id, remote_id, existing, direction, self.detect_established) { + if new_loses( + local_id, + remote_id, + existing, + direction, + self.detect_established, + ) { COLLISIONS.with_label_values(&["new"]).inc(); return None; } @@ -102,8 +121,21 @@ impl CollisionTable { } let id = self.next_id.fetch_add(1, Ordering::Relaxed); let (kill, killed) = oneshot::channel(); - entries.insert(remote_id, Entry { id, direction, established: false, kill: Some(kill) }); - Some(ConnectionGuard { table: self.clone(), remote_id, id, killed }) + entries.insert( + remote_id, + Entry { + id, + direction, + established: false, + kill: Some(kill), + }, + ); + Some(ConnectionGuard { + table: self.clone(), + remote_id, + id, + killed, + }) } } @@ -131,7 +163,10 @@ impl Drop for ConnectionGuard { // 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"); - if entries.get(&self.remote_id).is_some_and(|e| e.id == self.id) { + if entries + .get(&self.remote_id) + .is_some_and(|e| e.id == self.id) + { entries.remove(&self.remote_id); } } @@ -145,8 +180,20 @@ mod tests { const LOW: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 1); const HIGH: Ipv4Addr = Ipv4Addr::new(192, 0, 2, 1); - fn loses(local: Ipv4Addr, remote: Ipv4Addr, existing: Direction, established: bool, new: Direction, detect: bool) -> bool { - let e = Entry { id: 0, direction: existing, established, kill: None }; + fn loses( + local: Ipv4Addr, + remote: Ipv4Addr, + existing: Direction, + established: bool, + new: Direction, + detect: bool, + ) -> bool { + let e = Entry { + id: 0, + direction: existing, + established, + kill: None, + }; new_loses(local, remote, &e, new, detect) } @@ -169,7 +216,10 @@ mod tests { for (a, b) in [(LOW, HIGH), (HIGH, LOW)] { let a_closes_dial_from_b = loses(a, b, Outbound, false, Inbound, false); let b_closes_dial_from_a = loses(b, a, Outbound, false, Inbound, false); - assert_ne!(a_closes_dial_from_b, b_closes_dial_from_a, "ends disagree for a={a} b={b}"); + assert_ne!( + a_closes_dial_from_b, b_closes_dial_from_a, + "ends disagree for a={a} b={b}" + ); } } @@ -206,7 +256,11 @@ mod tests { assert!(table.admit(LOW, HIGH, Outbound).is_none()); // A different identifier never collides. - assert!(table.admit(LOW, Ipv4Addr::new(192, 0, 2, 2), Inbound).is_some()); + assert!( + table + .admit(LOW, Ipv4Addr::new(192, 0, 2, 2), Inbound) + .is_some() + ); // Once the survivor goes away the identifier is free again. drop(inb); diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index a8e4bd1..ceb145c 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -2,9 +2,9 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::str::FromStr; use std::time::Duration; -use bgp_primitives::ip_prefix::IpPrefix; use bgp_packet::community; use bgp_packet::message::OriginPathAttribute; +use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::{LocalRoute, LocalRouteAttrs}; use eyre::{Result, WrapErr, bail, eyre}; @@ -49,8 +49,12 @@ pub struct PeerConfig { impl Config { pub fn from_proto(proto: BgpConfig) -> Result { - let router_id = Ipv4Addr::from_str(&proto.router_id) - .map_err(|_| eyre!("Invalid router_id {:?}: must be a dotted-quad IPv4 address", proto.router_id))?; + let router_id = Ipv4Addr::from_str(&proto.router_id).map_err(|_| { + eyre!( + "Invalid router_id {:?}: must be a dotted-quad IPv4 address", + proto.router_id + ) + })?; if proto.local_asn == 0 { bail!("local_asn must be non-zero"); @@ -59,10 +63,14 @@ impl Config { let listen_endpoints = if proto.listen_endpoints.is_empty() { vec![ SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), DEFAULT_BGP_PORT), - SocketAddr::new(IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED), DEFAULT_BGP_PORT), + SocketAddr::new( + IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED), + DEFAULT_BGP_PORT, + ), ] } else { - proto.listen_endpoints + proto + .listen_endpoints .into_iter() .map(|ep| parse_endpoint(&ep.address, ep.port)) .collect::>()? @@ -108,10 +116,16 @@ impl Config { }) .collect::>()?; - let originate: Vec = - proto.originate.iter().map(parse_originated_route).collect::>()?; + let originate: Vec = proto + .originate + .iter() + .map(parse_originated_route) + .collect::>()?; for (i, route) in originate.iter().enumerate() { - if originate[..i].iter().any(|earlier| earlier.prefix == route.prefix) { + if originate[..i] + .iter() + .any(|earlier| earlier.prefix == route.prefix) + { bail!("originate lists {} more than once", route.prefix); } } @@ -155,8 +169,7 @@ fn parse_endpoint(address: &str, port: u32) -> Result { port as u16 }; - let ip = IpAddr::from_str(address) - .map_err(|_| eyre!("Invalid IP address {address:?}"))?; + let ip = IpAddr::from_str(address).map_err(|_| eyre!("Invalid IP address {address:?}"))?; Ok(SocketAddr::new(ip, port)) } @@ -202,7 +215,8 @@ mod tests { use super::*; fn parse(text: &str) -> Result { - let proto = protobuf::text_format::parse_from_str::(text).map_err(|e| eyre!("{e}"))?; + let proto = + protobuf::text_format::parse_from_str::(text).map_err(|e| eyre!("{e}"))?; Config::from_proto(proto) } @@ -228,7 +242,11 @@ mod tests { let full = &config.originate[1].attrs; assert_eq!(full.origin, OriginPathAttribute::INCOMPLETE); assert_eq!(full.prepend, 2); - assert_eq!(full.med, Some(0), "an explicit MED of 0 is not the same as no MED"); + assert_eq!( + full.med, + Some(0), + "an explicit MED of 0 is not the same as no MED" + ); assert_eq!(full.communities, vec![(65000, 100), (65535, 65281)]); assert_eq!(full.large_communities.len(), 1); } @@ -237,12 +255,23 @@ mod tests { fn bad_originated_routes_are_rejected_with_the_prefix_named() { for (body, expected) in [ ("prefix: \"not-a-prefix\"", "Invalid originate prefix"), - ("prefix: \"192.0.2.0/24\" origin: \"static\"", "must be IGP, EGP or INCOMPLETE"), + ( + "prefix: \"192.0.2.0/24\" origin: \"static\"", + "must be IGP, EGP or INCOMPLETE", + ), ("prefix: \"192.0.2.0/24\" as_path_prepend: 17", "at most 16"), - ("prefix: \"192.0.2.0/24\" communities: \"70000:1\"", "originate 192.0.2.0/24"), - ("prefix: \"192.0.2.0/24\" large_communities: \"1:2\"", "originate 192.0.2.0/24"), + ( + "prefix: \"192.0.2.0/24\" communities: \"70000:1\"", + "originate 192.0.2.0/24", + ), + ( + "prefix: \"192.0.2.0/24\" large_communities: \"1:2\"", + "originate 192.0.2.0/24", + ), ] { - let err = parse(&format!("{BASE}originate {{ {body} }}")).err().expect(body); + let err = parse(&format!("{BASE}originate {{ {body} }}")) + .err() + .expect(body); assert!(format!("{err:#}").contains(expected), "{body}: {err:#}"); } let err = parse(&format!( diff --git a/crates/server/src/dialer.rs b/crates/server/src/dialer.rs index 7a51b6a..151f435 100644 --- a/crates/server/src/dialer.rs +++ b/crates/server/src/dialer.rs @@ -49,10 +49,16 @@ async fn connect(peer: &PeerConfig) -> std::io::Result { let Some(local) = peer.local_address else { return TcpStream::connect(peer.endpoint).await; }; - let socket = if peer.endpoint.is_ipv4() { TcpSocket::new_v4()? } else { TcpSocket::new_v6()? }; + let socket = if peer.endpoint.is_ipv4() { + TcpSocket::new_v4()? + } else { + TcpSocket::new_v6()? + }; socket .bind(std::net::SocketAddr::new(local, 0)) - .map_err(|e| std::io::Error::new(e.kind(), format!("cannot use local_address {local}: {e}")))?; + .map_err(|e| { + std::io::Error::new(e.kind(), format!("cannot use local_address {local}: {e}")) + })?; socket.connect(peer.endpoint).await } @@ -78,7 +84,9 @@ async fn dial_loop(peer: Arc, config: SessionConfig, connect_retry: Ok(Ok(stream)) => { info!(peer = %peer.name, endpoint = %peer.endpoint, "TCP connected, starting session"); backoff = connect_retry; - run_session(stream, peer.endpoint, config.clone()).await.map(|()| "session closed by peer".to_string()) + run_session(stream, peer.endpoint, config.clone()) + .await + .map(|()| "session closed by peer".to_string()) } Ok(Err(e)) => Err(eyre::eyre!("connect failed: {e}")), Err(_) => Err(eyre::eyre!("connect timed out after {CONNECT_TIMEOUT:?}")), @@ -94,7 +102,10 @@ async fn dial_loop(peer: Arc, config: SessionConfig, connect_retry: registry.set_dial_state( &peer.name, - DialState::Backoff { until: SystemTime::now() + backoff, last_error: reason }, + DialState::Backoff { + until: SystemTime::now() + backoff, + last_error: reason, + }, ); tokio::time::sleep(backoff).await; backoff = (backoff * 2).min(MAX_BACKOFF); diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index a7a404f..42e09ea 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -19,7 +19,9 @@ use tracing::{info, warn}; use bgp_packet::message::{AS_TRANS, BGP4_VERSION}; use crate::collision::{CollisionTable, Direction}; -use crate::metrics::{MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, message_type_name}; +use crate::metrics::{ + MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, message_type_name, +}; use crate::session::{SessionHandle, SessionRegistry}; /// Number of outbound messages that may be queued for a session before @@ -125,7 +127,12 @@ pub async fn run_session( if let Some(expected) = config.expected_asn && peer_asn != expected { - send_notification(&mut writer, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), &ctx).await; + send_notification( + &mut writer, + BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), + &ctx, + ) + .await; bail!("Peer {peer_addr} announced AS{peer_asn} but AS{expected} was configured"); } @@ -133,7 +140,9 @@ pub async fn run_session( // guard keeps this connection visible to later ones, and `killed` fires // if one of them wins. let Some(mut collision_guard) = - config.collisions.admit(config.router_id, peer_open.identifier, config.direction) + config + .collisions + .admit(config.router_id, peer_open.identifier, config.direction) else { return close_for_collision(&mut writer, &ctx, peer_addr).await; }; @@ -152,7 +161,11 @@ pub async fn run_session( // With a negotiated hold time of zero the hold timer does not run // (RFC 4271 §4.2); fall back to the large pre-negotiation value so a // silent peer is still cleaned up eventually. - let hold_duration = Duration::from_secs(if negotiated_hold_time == 0 { 240 } else { negotiated_hold_time as u64 }); + let hold_duration = Duration::from_secs(if negotiated_hold_time == 0 { + 240 + } else { + negotiated_hold_time as u64 + }); let confirm = tokio::select! { msg = recv_with_timeout(&mut reader, &ctx, hold_duration) => msg, @@ -181,7 +194,10 @@ pub async fn run_session( info!(%peer_addr, peer_asn, "Session established"); SESSIONS_ESTABLISHED_TOTAL.inc(); - let peer_id = PeerId { addr: peer_addr, asn: peer_asn }; + let peer_id = PeerId { + addr: peer_addr, + asn: peer_asn, + }; // --- Established -------------------------------------------------------- // Register so the rest of the server can see this session and queue @@ -204,7 +220,11 @@ pub async fn run_session( negotiated_hold_time, peer_id, &config.adj_rib_in, - SessionInputs { outbound: &mut outbound_rx, killed: &mut collision_guard.killed, registry: &config.registry }, + SessionInputs { + outbound: &mut outbound_rx, + killed: &mut collision_guard.killed, + registry: &config.registry, + }, ) .await; @@ -235,12 +255,20 @@ async fn established_loop( adj_rib_in: &AdjRibIn, inputs: SessionInputs<'_>, ) -> Result<()> { - let SessionInputs { outbound, killed, registry } = 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. let timers_enabled = hold_time_secs != 0; - let secs = if timers_enabled { hold_time_secs as u64 } else { 3 }; + let secs = if timers_enabled { + hold_time_secs as u64 + } else { + 3 + }; let hold_duration = Duration::from_secs(secs); let keepalive_interval = Duration::from_secs(secs / 3); @@ -328,7 +356,9 @@ async fn read_message(reader: &mut OwnedReadHalf, ctx: &ParserContext) -> Result let (_, msg) = Message::from_wire(ctx, &buf).map_err(|e| eyre!("Failed to parse BGP message: {e:?}"))?; - MESSAGES_RECEIVED.with_label_values(&[message_type_name(&msg)]).inc(); + MESSAGES_RECEIVED + .with_label_values(&[message_type_name(&msg)]) + .inc(); Ok(msg) } @@ -350,7 +380,9 @@ async fn send_message( let mut buf = BytesMut::new(); msg.to_wire(ctx, &mut buf)?; writer.write_all(&buf).await?; - MESSAGES_SENT.with_label_values(&[message_type_name(msg)]).inc(); + MESSAGES_SENT + .with_label_values(&[message_type_name(msg)]) + .inc(); Ok(()) } @@ -374,7 +406,12 @@ async fn close_for_collision( peer_addr: SocketAddr, ) -> Result<()> { info!(%peer_addr, "Closing connection: lost collision resolution"); - send_notification(writer, BgpError::Cease(CeaseSubcode::ConnCollisionResolution), ctx).await; + send_notification( + writer, + BgpError::Cease(CeaseSubcode::ConnCollisionResolution), + ctx, + ) + .await; bail!("Connection to {peer_addr} closed by collision resolution"); } @@ -413,7 +450,12 @@ async fn apply_update( rib.withdraw(peer, prefix.clone()).await; } for prefix in announced { - rib.announce(peer, prefix.clone(), RibEntry::new(path_attributes.to_vec())).await; + rib.announce( + peer, + prefix.clone(), + RibEntry::new(path_attributes.to_vec()), + ) + .await; } // Multiprotocol attributes carry NLRI for all other AFIs. @@ -421,7 +463,12 @@ async fn apply_update( match attr { PathAttribute::MpReachNlri(mp) => { for prefix in &mp.prefixes { - rib.announce(peer, prefix.clone(), RibEntry::new(path_attributes.to_vec())).await; + rib.announce( + peer, + prefix.clone(), + RibEntry::new(path_attributes.to_vec()), + ) + .await; } } PathAttribute::MpUnreachNlri(mp) => { diff --git a/crates/server/src/metrics.rs b/crates/server/src/metrics.rs index 37ce935..1b1fe08 100644 --- a/crates/server/src/metrics.rs +++ b/crates/server/src/metrics.rs @@ -9,23 +9,37 @@ use prometheus::{ }; pub static MESSAGES_SENT: LazyLock = LazyLock::new(|| { - register_int_counter_vec!("bgp_messages_sent_total", "BGP messages sent, by type", &["type"]) - .expect("metric registration") + register_int_counter_vec!( + "bgp_messages_sent_total", + "BGP messages sent, by type", + &["type"] + ) + .expect("metric registration") }); pub static MESSAGES_RECEIVED: LazyLock = LazyLock::new(|| { - register_int_counter_vec!("bgp_messages_received_total", "BGP messages received, by type", &["type"]) - .expect("metric registration") + register_int_counter_vec!( + "bgp_messages_received_total", + "BGP messages received, by type", + &["type"] + ) + .expect("metric registration") }); pub static SESSIONS_ESTABLISHED_TOTAL: LazyLock = LazyLock::new(|| { - register_int_counter!("bgp_sessions_established_total", "Sessions that reached Established") - .expect("metric registration") + register_int_counter!( + "bgp_sessions_established_total", + "Sessions that reached Established" + ) + .expect("metric registration") }); pub static SESSIONS_ESTABLISHED: LazyLock = LazyLock::new(|| { - register_int_gauge!("bgp_sessions_established", "Sessions currently in Established") - .expect("metric registration") + register_int_gauge!( + "bgp_sessions_established", + "Sessions currently in Established" + ) + .expect("metric registration") }); pub static COLLISIONS: LazyLock = LazyLock::new(|| { @@ -38,12 +52,16 @@ pub static COLLISIONS: LazyLock = LazyLock::new(|| { }); pub static LOCAL_ROUTES: LazyLock = LazyLock::new(|| { - register_int_gauge!("bgp_local_routes", "Locally originated routes").expect("metric registration") + register_int_gauge!("bgp_local_routes", "Locally originated routes") + .expect("metric registration") }); pub static ADJ_RIB_IN_ROUTES: LazyLock = LazyLock::new(|| { - register_int_gauge!("bgp_adj_rib_in_routes", "Routes held in Adj-RIB-In across all peers") - .expect("metric registration") + register_int_gauge!( + "bgp_adj_rib_in_routes", + "Routes held in Adj-RIB-In across all peers" + ) + .expect("metric registration") }); /// Forces registration of every metric so they appear in scrapes before diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index fb2522d..a96082c 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -57,7 +57,10 @@ 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.wrap_err("Local RIB stopped during startup")?; + local_rib + .announce(route.clone()) + .await + .wrap_err("Local RIB stopped during startup")?; } Ok(BgpServer { listeners, @@ -67,7 +70,11 @@ impl BgpServerBuilder { self.config .peers .iter() - .map(|p| ConfiguredPeer { name: p.name.clone(), endpoint: p.endpoint, remote_asn: p.remote_asn }) + .map(|p| ConfiguredPeer { + name: p.name.clone(), + endpoint: p.endpoint, + remote_asn: p.remote_asn, + }) .collect(), ), config: self.config, @@ -116,7 +123,11 @@ impl BgpServer { self.registry.clone(), self.config.local_asn, ); - let _dialers = spawn_dialers(&self.config.peers, &session_config, self.config.connect_retry); + let _dialers = spawn_dialers( + &self.config.peers, + &session_config, + self.config.connect_retry, + ); let mut accepting: FuturesUnordered<_> = self .listeners @@ -130,7 +141,12 @@ impl BgpServer { // Inbound connections from a configured peer inherit its // name, expected ASN and hold time. let mut config = session_config.clone(); - if let Some(peer) = self.config.peers.iter().find(|p| p.endpoint.ip() == peer_addr.ip()) { + if let Some(peer) = self + .config + .peers + .iter() + .find(|p| p.endpoint.ip() == peer_addr.ip()) + { config.peer_name = Some(peer.name.clone()); config.expected_asn = Some(peer.remote_asn); config.hold_time_secs = peer.hold_time_secs; diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index b20b840..49f0f77 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -30,7 +30,10 @@ pub enum DialState { Idle, Connecting, /// The last attempt failed or the session ended; waiting to retry. - Backoff { until: SystemTime, last_error: String }, + Backoff { + until: SystemTime, + last_error: String, + }, } /// A peer from the configuration file, whether or not it is currently up. @@ -72,13 +75,24 @@ impl SessionHandle { four_octet_asn: bool, outbound: mpsc::Sender, ) -> Self { - Self { peer, name, direction, local_addr, established_at: SystemTime::now(), four_octet_asn, outbound } + Self { + peer, + name, + direction, + local_addr, + established_at: SystemTime::now(), + four_octet_asn, + outbound, + } } /// Queues a message for transmission on this session. Waits for queue /// space if the session's writer is behind. pub async fn send(&self, msg: Message) -> Result<(), SessionClosed> { - self.outbound.send(msg).await.map_err(|_| SessionClosed(self.peer)) + self.outbound + .send(msg) + .await + .map_err(|_| SessionClosed(self.peer)) } } @@ -114,7 +128,10 @@ impl SessionRegistry { pub fn with_configured(configured: Vec) -> Self { let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); - let dial_states = configured.iter().map(|p| (p.name.clone(), DialState::Idle)).collect(); + let dial_states = configured + .iter() + .map(|p| (p.name.clone(), DialState::Idle)) + .collect(); Self { inner: Arc::new(RwLock::new(HashMap::new())), configured: Arc::new(configured), @@ -130,7 +147,12 @@ impl SessionRegistry { let states = self.dial_states.read().expect("dial states poisoned"); self.configured .iter() - .map(|p| (p.clone(), states.get(&p.name).cloned().unwrap_or(DialState::Idle))) + .map(|p| { + ( + p.clone(), + states.get(&p.name).cloned().unwrap_or(DialState::Idle), + ) + }) .collect() } @@ -140,7 +162,10 @@ 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 + .write() + .expect("dial states poisoned") + .insert(name.to_string(), state); } /// An established session with this remote IP and ASN, regardless of @@ -158,19 +183,34 @@ 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 + .write() + .expect("session registry poisoned") + .insert(peer, handle.clone()); SESSIONS_ESTABLISHED.inc(); let _ = self.events.send(SessionEvent::Up(handle)); - SessionGuard { registry: self.clone(), peer } + SessionGuard { + registry: self.clone(), + peer, + } } pub fn get(&self, peer: &PeerId) -> Option { - self.inner.read().expect("session registry poisoned").get(peer).cloned() + self.inner + .read() + .expect("session registry poisoned") + .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 + .read() + .expect("session registry poisoned") + .values() + .cloned() + .collect() } /// Called by a session when its peer asks for a route refresh. @@ -212,7 +252,11 @@ impl SessionRegistry { } fn remove(&self, peer: &PeerId) { - let removed = self.inner.write().expect("session registry poisoned").remove(peer); + let removed = self + .inner + .write() + .expect("session registry poisoned") + .remove(peer); if removed.is_some() { SESSIONS_ESTABLISHED.dec(); let _ = self.events.send(SessionEvent::Down(*peer)); diff --git a/crates/web/src/api.rs b/crates/web/src/api.rs index 29a6d78..82eaf29 100644 --- a/crates/web/src/api.rs +++ b/crates/web/src/api.rs @@ -18,7 +18,9 @@ use crate::AppState; // --------------------------------------------------------------------------- fn unix_secs(t: SystemTime) -> u64 { - t.duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0) + t.duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) } #[derive(Debug, Serialize)] @@ -76,7 +78,9 @@ impl PeerView { let (state_name, last_error, retry_at_unix) = match state { DialState::Idle => ("Idle", None, None), DialState::Connecting => ("Connecting", None, None), - DialState::Backoff { until, last_error } => ("Backoff", Some(last_error.clone()), Some(unix_secs(*until))), + DialState::Backoff { until, last_error } => { + ("Backoff", Some(last_error.clone()), Some(unix_secs(*until))) + } }; Self { name: Some(peer.name.clone()), @@ -103,17 +107,26 @@ pub fn peer_views(registry: &SessionRegistry) -> Vec { .configured_peers() .iter() .map(|(peer, state)| { - match sessions.iter().find(|s| s.peer.addr.ip() == peer.endpoint.ip() && s.peer.asn == peer.remote_asn) { + match sessions + .iter() + .find(|s| s.peer.addr.ip() == peer.endpoint.ip() && s.peer.asn == peer.remote_asn) + { Some(session) => { claimed.insert(session.peer); - PeerView { name: Some(peer.name.clone()), ..PeerView::from(session) } + PeerView { + name: Some(peer.name.clone()), + ..PeerView::from(session) + } } None => PeerView::from_configured(peer, state), } }) .collect(); - let mut rest: Vec = - sessions.iter().filter(|s| !claimed.contains(&s.peer)).map(PeerView::from).collect(); + let mut rest: Vec = sessions + .iter() + .filter(|s| !claimed.contains(&s.peer)) + .map(PeerView::from) + .collect(); rest.sort_by(|a, b| a.peer_addr.cmp(&b.peer_addr)); views.extend(rest); views @@ -137,8 +150,19 @@ impl From<&LocalRoute> for LocalRouteView { origin: origin_name(r.attrs.origin).to_string(), prepend: r.attrs.prepend, med: r.attrs.med, - communities: r.attrs.communities.iter().copied().map(community::format_standard).collect(), - large_communities: r.attrs.large_communities.iter().map(community::format_large).collect(), + communities: r + .attrs + .communities + .iter() + .copied() + .map(community::format_standard) + .collect(), + large_communities: r + .attrs + .large_communities + .iter() + .map(community::format_large) + .collect(), announced_at_unix: unix_secs(r.announced_at), } } @@ -163,14 +187,22 @@ pub fn summarize(attrs: &[PathAttribute]) -> AttrSummary { match attr { PathAttribute::Origin(o) => s.origin = Some(origin_name(*o).to_string()), PathAttribute::AsPath(p) => { - s.as_path = p.segments.iter().flat_map(|seg| seg.path.iter().copied()).collect() + s.as_path = p + .segments + .iter() + .flat_map(|seg| seg.path.iter().copied()) + .collect() } PathAttribute::NextHop(nh) => s.next_hop = Some(nh.0.to_string()), PathAttribute::MpReachNlri(mp) => s.next_hop = Some(format!("{:?}", mp.next_hop)), PathAttribute::MultiExitDisc(m) => s.med = Some(m.0), PathAttribute::LocalPref(l) => s.local_pref = Some(l.0), PathAttribute::Communities(c) => { - s.communities = c.0.iter().copied().map(community::format_standard).collect() + s.communities = + c.0.iter() + .copied() + .map(community::format_standard) + .collect() } PathAttribute::LargeCommunities(l) => { s.large_communities = l.communities.iter().map(community::format_large).collect() @@ -250,15 +282,25 @@ pub async fn peers(State(state): State) -> Json> { } pub async fn local_routes(State(state): State) -> Json> { - let mut routes: Vec = - state.local_rib.snapshot().await.iter().map(LocalRouteView::from).collect(); + let mut routes: Vec = state + .local_rib + .snapshot() + .await + .iter() + .map(LocalRouteView::from) + .collect(); routes.sort_by(|a, b| a.prefix.cmp(&b.prefix)); Json(routes) } pub async fn received_routes(State(state): State) -> Json> { - let mut routes: Vec = - state.adj_rib_in.snapshot().await.iter().map(ReceivedRouteView::from).collect(); + let mut routes: Vec = state + .adj_rib_in + .snapshot() + .await + .iter() + .map(ReceivedRouteView::from) + .collect(); routes.sort_by(|a, b| (&a.prefix, &a.peer_addr).cmp(&(&b.prefix, &b.peer_addr))); Json(routes) } diff --git a/crates/web/src/events.rs b/crates/web/src/events.rs index 6c20676..046d5ba 100644 --- a/crates/web/src/events.rs +++ b/crates/web/src/events.rs @@ -18,40 +18,81 @@ use crate::api::{LocalRouteView, PeerView, ReceivedRouteView, summarize}; #[derive(Debug, Serialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum WireEvent { - SessionUp { peer: PeerView }, - SessionDown { peer_addr: String, peer_asn: u32 }, + 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 }, - ReceivedWithdrawn { peer_addr: String, peer_asn: u32, prefix: String }, + RouteRefreshRequested { + peer: PeerView, + }, + LocalAnnounced { + route: LocalRouteView, + }, + LocalWithdrawn { + prefix: String, + }, + ReceivedAnnounced { + route: ReceivedRouteView, + }, + ReceivedWithdrawn { + peer_addr: String, + peer_asn: u32, + prefix: String, + }, /// The client fell behind and missed `missed` events on `stream`; it /// should refetch the corresponding table. - Lagged { stream: &'static str, missed: u64 }, + Lagged { + stream: &'static str, + missed: u64, + }, } impl WireEvent { fn from_session(ev: Result) -> Self { 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 }, + 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, + }, } } fn from_local(ev: Result) -> Self { match ev { - Ok(LocalRibEvent::Announced(r)) => Self::LocalAnnounced { route: LocalRouteView::from(&r) }, - Ok(LocalRibEvent::Withdrawn { prefix }) => Self::LocalWithdrawn { prefix: prefix.to_string() }, - Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged { stream: "local_routes", missed: n }, + Ok(LocalRibEvent::Announced(r)) => Self::LocalAnnounced { + route: LocalRouteView::from(&r), + }, + Ok(LocalRibEvent::Withdrawn { prefix }) => Self::LocalWithdrawn { + prefix: prefix.to_string(), + }, + Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged { + stream: "local_routes", + missed: n, + }, } } fn from_received(ev: Result) -> Self { match ev { - Ok(RibEvent::Announced { peer, prefix, entry }) => Self::ReceivedAnnounced { + Ok(RibEvent::Announced { + peer, + prefix, + entry, + }) => Self::ReceivedAnnounced { route: ReceivedRouteView { peer_addr: peer.addr.to_string(), peer_asn: peer.asn, @@ -70,7 +111,10 @@ impl WireEvent { peer_asn: peer.asn, prefix: prefix.to_string(), }, - Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged { stream: "received_routes", missed: n }, + Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged { + stream: "received_routes", + missed: n, + }, } } } @@ -82,12 +126,13 @@ pub async fn stream( let local = BroadcastStream::new(state.local_rib.subscribe()).map(WireEvent::from_local); let received = BroadcastStream::new(state.adj_rib_in.subscribe()).map(WireEvent::from_received); - let merged = futures::stream::select(sessions, futures::stream::select(local, received)).map(|ev| { - Ok(Event::default() - .event("bgp") - .json_data(&ev) - .unwrap_or_else(|e| Event::default().event("error").data(e.to_string()))) - }); + let merged = + futures::stream::select(sessions, futures::stream::select(local, received)).map(|ev| { + Ok(Event::default() + .event("bgp") + .json_data(&ev) + .unwrap_or_else(|e| Event::default().event("error").data(e.to_string()))) + }); Sse::new(merged).keep_alive(KeepAlive::default()) } diff --git a/crates/web/src/lib.rs b/crates/web/src/lib.rs index 5cbaa50..017e276 100644 --- a/crates/web/src/lib.rs +++ b/crates/web/src/lib.rs @@ -86,7 +86,10 @@ pub struct AppState { const STATUSZ_HTML: &str = include_str!("../static/statusz.html"); async fn statusz() -> impl IntoResponse { - ([(header::CONTENT_TYPE, "text/html; charset=utf-8")], Html(STATUSZ_HTML)) + ( + [(header::CONTENT_TYPE, "text/html; charset=utf-8")], + Html(STATUSZ_HTML), + ) } /// Builds the router. Exposed so tests can drive it without a socket. diff --git a/crates/web/src/metrics.rs b/crates/web/src/metrics.rs index cd0c6e6..4f7636f 100644 --- a/crates/web/src/metrics.rs +++ b/crates/web/src/metrics.rs @@ -15,7 +15,12 @@ pub async fn metricz(State(state): State) -> impl IntoResponse { let mut buf = Vec::new(); match TextEncoder::new().encode(&prometheus::gather(), &mut buf) { - Ok(()) => (StatusCode::OK, [(header::CONTENT_TYPE, "text/plain; version=0.0.4")], buf).into_response(), + Ok(()) => ( + StatusCode::OK, + [(header::CONTENT_TYPE, "text/plain; version=0.0.4")], + buf, + ) + .into_response(), Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(), } }