diff --git a/crates/fib/src/interfaces.rs b/crates/fib/src/interfaces.rs index 51386d4..4672534 100644 --- a/crates/fib/src/interfaces.rs +++ b/crates/fib/src/interfaces.rs @@ -2,12 +2,13 @@ //! them, and what they say about a next hop: whether it can be reached //! directly, and from which interface. -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeMap, HashSet}; use std::net::IpAddr; use bgp_primitives::constants::IpVersion; use bgp_primitives::ip_prefix::IpPrefix; +use crate::InterfaceStatus; use crate::netlink::{Address, Link, Message}; /// The interface a neighbor is reached on. @@ -22,7 +23,8 @@ pub struct Connected { /// The interfaces there are and the addresses they have. #[derive(Debug, Clone, Default)] pub struct Interfaces { - links: HashMap, + /// By index, which is the order they are given out in too. + links: BTreeMap, addresses: HashSet
, } @@ -64,11 +66,31 @@ impl Interfaces { } } - /// The interfaces, by index. + /// The interfaces, in order of index. pub fn links(&self) -> impl Iterator { self.links.values() } + /// The interfaces and their addresses, for showing: in order of index, + /// each with its addresses in order. + pub fn status(&self) -> Vec { + self.links() + .map(|link| { + let mut addresses: Vec<(IpAddr, u8)> = self + .addresses_of(link.index) + .map(|address| (address.local, address.network.length())) + .collect(); + addresses.sort(); + InterfaceStatus { + index: link.index, + name: link.name.clone(), + up: link.up, + addresses, + } + }) + .collect() + } + pub fn link(&self, index: u32) -> Option<&Link> { self.links.get(&index) } @@ -237,4 +259,28 @@ mod tests { // What is not about interfaces changes nothing. assert!(!interfaces.apply(&Message::Done)); } + + #[test] + fn the_interfaces_are_shown_in_order() { + let mut interfaces = some(); + // Learned last, listed in its place. + assert!(interfaces.apply(&link(0, "dummy0", false))); + let shown = interfaces.status(); + let indexes: Vec = shown.iter().map(|interface| interface.index).collect(); + assert_eq!(indexes, [0, 1, 2, 3, 4]); + assert_eq!( + shown[2], + InterfaceStatus { + index: 2, + name: "eth0".into(), + up: true, + addresses: vec![ + ("192.0.2.1".parse().unwrap(), 24), + ("2001:db8::1".parse().unwrap(), 64), + ], + } + ); + assert_eq!((shown[4].up, shown[4].addresses.len()), (false, 1)); + assert!(shown[0].addresses.is_empty()); + } } diff --git a/crates/fib/src/kernel.rs b/crates/fib/src/kernel.rs index 4278cb1..614eb46 100644 --- a/crates/fib/src/kernel.rs +++ b/crates/fib/src/kernel.rs @@ -3,7 +3,6 @@ use std::collections::{HashMap, HashSet}; use std::io; -use std::net::IpAddr; use std::sync::{Arc, LazyLock}; use std::time::Duration; @@ -19,7 +18,7 @@ use prometheus::{ register_int_gauge, }; use tokio::sync::{mpsc, watch}; -use tokio::task::JoinHandle; +use tokio::task::{JoinError, JoinHandle}; use tokio::time::{Instant, sleep_until}; use tracing::{debug, info, warn}; @@ -27,11 +26,10 @@ use crate::interfaces::Interfaces; use crate::netlink::{ self, Change, Dump, KIND_UNICAST, Message, PROTOCOL_BGP, Route, decode, group, }; - -/// "Input/output error": for what should not happen. -const EIO: i32 = 5; -use crate::plan::{Neighbor, Op, Plan, Target, Unresolved, resolve}; -use crate::{Fib, FibConfig, FibError, InterfaceStatus, Stop}; +use crate::plan::{ + HashMap as FastHashMap, Neighbor, Op, Plan, Target, Unresolved, resolve_next_hop, +}; +use crate::{Fib, FibConfig, FibError, Stop}; /// How many best routes are taken from the routing table at a time. const BATCH: usize = 1024; @@ -77,6 +75,8 @@ mod errno { pub const EPERM: i32 = 1; pub const ENOENT: i32 = 2; pub const ESRCH: i32 = 3; + /// "Input/output error": for what should not happen. + pub const EIO: i32 = 5; pub const ENOBUFS: i32 = 105; } @@ -438,12 +438,12 @@ struct Task { /// share them; looking through the interfaces for each route would /// take as long as there are addresses. Emptied when the interfaces /// change. - resolved: HashMap, foldhash::fast::RandomState>, + resolved: FastHashMap>, plan: Plan, /// The kernel's nexthop objects that our routes name, by where they /// lead; `None` if routes carry their own gateway instead, because /// the kernel has no such objects or the configuration says so. - nexthops: Option>, + nexthops: Option>, /// The numbers of the nexthop objects that are ours: made by this run, /// or named by routes of ours that an earlier run left. Others of /// BGP's are another daemon's, and are neither used nor removed: @@ -620,7 +620,7 @@ impl Task { Ok(id) } // Made, and the kernel did not say under what number. - (0, None) => Err(EIO), + (0, None) => Err(errno::EIO), (errno, _) => Err(errno), }) } @@ -693,18 +693,16 @@ impl Task { let queue = self.queue.as_ref().expect("nothing is sent after the end"); if queue.send(batch).await.is_err() { // The writer has ended, which it does only by failing. - return Err(self.writer_failed().await); + return Err(writer_error((&mut self.writer).await)); } Ok(()) } - /// Why the task that writes to the kernel ended. - async fn writer_failed(&mut self) -> FibError { - match (&mut self.writer).await { - Ok(Err(e)) => e, - Ok(Ok(())) => io::Error::other("the task that writes to the kernel ended").into(), - Err(e) => io::Error::other(e).into(), - } + /// Lets the writer know there is no more to come, and waits until it + /// has written what there is. + async fn finish_writer(&mut self) -> Result<(), FibError> { + self.queue = None; + (&mut self.writer).await.map_err(io::Error::other)? } /// Deals with the kernel having refused `op`. Returns what to ask of @@ -729,18 +727,21 @@ impl Task { ), Op::Remove { prefix } => format!("remove {prefix}"), }; - // Counted always, said once in a while: a table's worth of - // refusals is a table's worth of lines. + // Counted always, said out loud once in a while: a table's worth + // of refusals is a table's worth of lines. ERRORS.inc(); let said = format!("the kernel refused to {what}: {error}"); - debug!("{said}"); - self.fib.update(|status| { - if status.errors == 0 || status.errors.is_power_of_two() { - warn!(refusals = status.errors + 1, "{said}"); - } + let before = self.fib.update(|status| { + let before = status.errors; status.errors += 1; - status.last_error = Some(said); + status.last_error = Some(said.clone()); + before }); + if before == 0 || before.is_power_of_two() { + warn!(refusals = before + 1, "{said}"); + } else { + debug!(refusals = before + 1, "{said}"); + } Ok(self.plan.refused_by_kernel(op)) } @@ -808,7 +809,12 @@ impl Task { .resolved .entry((next_hop, prefix.address_family(), peer)) .or_insert_with(|| { - resolve(interfaces, &prefix, next_hop.as_ref(), peer) + resolve_next_hop( + interfaces, + prefix.address_family(), + next_hop.as_ref(), + peer, + ) }) }; ops.extend(self.plan.best(prefix, target)); @@ -926,7 +932,7 @@ impl Task { Some(report) => { self.hear(report).await?; } - None => return Err(self.writer_failed().await), + None => return Err(writer_error((&mut self.writer).await)), } } Ok(()) @@ -939,9 +945,7 @@ impl Task { routes = self.plan.installed(), "Leaving our routes in the kernel for the next run" ); - self.queue = None; - (&mut self.writer).await.map_err(io::Error::other)??; - Ok(()) + self.finish_writer().await } /// Takes everything of ours out of the kernel. @@ -976,11 +980,7 @@ impl Task { } let ops = self.plan.clear(); self.send(ops, false).await?; - // No more to come: the writer ends when it has written what there - // is. - self.queue = None; - (&mut self.writer).await.map_err(io::Error::other)??; - Ok(()) + self.finish_writer().await } /// Says how things stand, for whoever asks. @@ -990,26 +990,7 @@ impl Task { REFUSED.set(self.plan.refused() as i64); let interfaces = interfaces_changed.then(|| { self.fib.publish_interfaces(&self.interfaces); - let mut interfaces: Vec = self - .interfaces - .links() - .map(|link| { - let mut addresses: Vec<(IpAddr, u8)> = self - .interfaces - .addresses_of(link.index) - .map(|address| (address.local, address.network.length())) - .collect(); - addresses.sort(); - InterfaceStatus { - index: link.index, - name: link.name.clone(), - up: link.up, - addresses, - } - }) - .collect(); - interfaces.sort_by_key(|interface| interface.index); - interfaces + self.interfaces.status() }); self.fib.update(|status| { status.synced = self.synced; @@ -1028,6 +1009,16 @@ impl Task { } } +/// Why the task that writes to the kernel ended, from what joining it +/// gave. It ends of its own accord only by failing. +fn writer_error(joined: Result, JoinError>) -> FibError { + match joined { + Ok(Err(e)) => e, + Ok(Ok(())) => io::Error::other("the task that writes to the kernel ended").into(), + Err(e) => io::Error::other(e).into(), + } +} + /// Keeps the kernel's routes in step with the best routes of `rib` until /// told to stop, and then takes them out. pub(crate) async fn run( @@ -1058,9 +1049,9 @@ pub(crate) async fn run( queue: Some(queue), writer: tokio::spawn(write(routes, batches, reports)), interfaces: Interfaces::default(), - resolved: HashMap::default(), + resolved: FastHashMap::default(), plan: Plan::default(), - nexthops: config.nexthop_objects.then(HashMap::default), + nexthops: config.nexthop_objects.then(FastHashMap::default), owned: HashSet::new(), awaiting: None, marks: 0, @@ -1148,13 +1139,7 @@ pub(crate) async fn run( task.send(stale, false).await?; task.publish(false); } - ended = &mut task.writer => { - return Err(match ended { - Ok(Err(e)) => e, - Ok(Ok(())) => io::Error::other("the task that writes to the kernel ended").into(), - Err(e) => io::Error::other(e).into(), - }); - } + ended = &mut task.writer => return Err(writer_error(ended)), _ = stopping.changed() => { if *stopping.borrow() == Some(Stop::Keep) { task.leave().await?; diff --git a/crates/fib/src/lib.rs b/crates/fib/src/lib.rs index a0cb2e4..eddcbb7 100644 --- a/crates/fib/src/lib.rs +++ b/crates/fib/src/lib.rs @@ -171,7 +171,8 @@ pub struct Fib { impl Fib { /// Starts keeping the kernel's routes in step with the best routes of /// `rib`, and returns a handle with the task that does it. The task - /// ends only by failing, or when [`Fib::stop`] is called. + /// ends only by failing, or when [`Fib::stop`] or [`Fib::leave`] is + /// called. /// /// Must be called from within a Tokio runtime. Fails at once where /// there is no kernel to talk to in this way. @@ -270,15 +271,16 @@ impl Fib { self.shared.release.send_replace(true); } + /// Changes the status, and gives back what `change` does. #[cfg_attr(not(target_os = "linux"), allow(dead_code))] - fn update(&self, change: impl FnOnce(&mut Status)) { + fn update(&self, change: impl FnOnce(&mut Status) -> R) -> R { change( &mut self .shared .status .lock() .unwrap_or_else(PoisonError::into_inner), - ); + ) } } diff --git a/crates/fib/src/netlink.rs b/crates/fib/src/netlink.rs index e40bc99..c30abf4 100644 --- a/crates/fib/src/netlink.rs +++ b/crates/fib/src/netlink.rs @@ -1091,9 +1091,14 @@ mod tests { } /// Noise behind a header that could be a message's, to get past it. + /// The types of nexthop messages lie far from the rest. #[test] fn noise_behind_a_plausible_header_is_never_more_than_an_error( - message_type in 0u16..30, + message_type in prop_oneof![ + 0u16..30, + Just(message_type::NEW_NEXTHOP), + Just(message_type::DEL_NEXTHOP), + ], body in proptest::collection::vec(any::(), 0..100), ) { let wire = message(message_type, 1, &body, &[]); diff --git a/crates/fib/src/plan.rs b/crates/fib/src/plan.rs index 66a383a..1a142fc 100644 --- a/crates/fib/src/plan.rs +++ b/crates/fib/src/plan.rs @@ -12,8 +12,8 @@ use bgp_primitives::ip_prefix::IpPrefix; use crate::interfaces::Interfaces; // Asked for every route: a quick hash matters here. -type HashMap = std::collections::HashMap; -type HashSet = std::collections::HashSet; +pub(crate) type HashMap = std::collections::HashMap; +pub(crate) type HashSet = std::collections::HashSet; /// Where the kernel is to send packets for a prefix. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -63,6 +63,16 @@ impl Unresolved { } } +// A reason's place in `ALL` is its number as declared, which is what +// [`Plan`] counts them by. +const _: () = { + let mut at = 0; + while at < Unresolved::ALL.len() { + assert!(Unresolved::ALL[at] as usize == at); + at += 1; + } +}; + /// The peer a route was learned from, as far as its next hop is concerned. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct Neighbor { @@ -259,12 +269,19 @@ impl Plan { Some(reason) => self.unresolved.insert(prefix, reason), None => self.unresolved.remove(&prefix), }; - for (reason, by) in [(before, -1isize), (reason, 1)] { - if let Some(reason) = reason { - let count = &mut self.unresolved_counts[reason as usize]; - *count = count.wrapping_add_signed(by); - } + if let Some(before) = before { + self.unresolved_counts[before as usize] -= 1; } + if let Some(reason) = reason { + self.unresolved_counts[reason as usize] += 1; + } + } + + /// Forgets which prefixes are unresolved or refused, and why. + fn forget_unresolved_and_refused(&mut self) { + self.unresolved.clear(); + self.unresolved_counts = [0; Unresolved::ALL.len()]; + self.refused.clear(); } /// How many prefixes have a best route that the kernel refused. @@ -272,11 +289,6 @@ impl Plan { self.refused.len() } - /// The prefixes of the routes the kernel has. - pub fn installed_prefixes(&self) -> impl Iterator { - self.installed.keys() - } - /// Notes a route of ours that the kernel already has. It stays only if /// the routing table turns out to have the same. /// @@ -326,9 +338,7 @@ impl Plan { for entry in self.installed.values_mut() { entry.confirmed = false; } - self.unresolved.clear(); - self.unresolved_counts = [0; Unresolved::ALL.len()]; - self.refused.clear(); + self.forget_unresolved_and_refused(); } /// The best route for `prefix` is now one that resolves as `target`. @@ -412,9 +422,7 @@ impl Plan { /// Everything goes: the changes that empty the kernel of our routes. pub fn clear(&mut self) -> Vec { - self.unresolved.clear(); - self.unresolved_counts = [0; Unresolved::ALL.len()]; - self.refused.clear(); + self.forget_unresolved_and_refused(); self.installed .drain() .map(|(prefix, _)| Op::Remove { prefix }) diff --git a/crates/rpki/src/client.rs b/crates/rpki/src/client.rs index ef31d69..d9da4e4 100644 --- a/crates/rpki/src/client.rs +++ b/crates/rpki/src/client.rs @@ -17,7 +17,6 @@ //! worse: it would go on refusing routes whose holders have since //! authorised them. -use std::collections::HashSet; use std::io; use std::net::SocketAddr; use std::sync::atomic::{AtomicU64, Ordering}; @@ -37,6 +36,10 @@ use crate::aspa::AspaTable; use crate::rtr::{Pdu, RtrError, Timers, VERSION_0, VERSION_2, error_code}; use crate::vrp::{Vrp, VrpTable}; +// Hashed by the hundred thousand with every answer, and told apart from +// what there was with every reset: a quick hash matters here. +type HashSet = std::collections::HashSet; + const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); /// How long a cache may take to send the next part of an answer. @@ -226,8 +229,14 @@ enum SessionError { Malformed(#[from] RtrError), #[error("the cache closed the connection")] Closed, - #[error("the cache did not answer within {RESPONSE_TIMEOUT:?}")] - Timeout, + #[error("the cache did not accept the connection within {0:?}")] + ConnectTimeout(Duration), + #[error("the cache sent nothing for {0:?} in the middle of an answer")] + ResponseTimeout(Duration), + #[error("the cache did not finish its answer within {0:?}")] + AnswerTimeout(Duration), + #[error("the cache confirmed nothing for so long that the data expired")] + Expired, #[error("the cache reported an error: {text} (code {code})")] Reported { code: u16, text: String }, #[error("the cache sent {0} where it has no place")] @@ -249,8 +258,14 @@ impl SessionError { Self::Malformed(e) => Some(e.code()), Self::Unexpected(_) => Some(error_code::CORRUPT_DATA), Self::Inconsistent(problem) => Some(problem.code), - Self::Io(_) | Self::Closed | Self::Timeout | Self::Reported { .. } => None, - Self::Downgrade(_) => None, + Self::Io(_) + | Self::Closed + | Self::ConnectTimeout(_) + | Self::ResponseTimeout(_) + | Self::AnswerTimeout(_) + | Self::Expired + | Self::Reported { .. } + | Self::Downgrade(_) => None, } } } @@ -488,17 +503,14 @@ impl Client { let stream = match timeout(CONNECT_TIMEOUT, TcpStream::connect(cache)).await { Ok(Ok(stream)) => stream, Ok(Err(e)) => return e.into(), - Err(_) => return SessionError::Timeout, + Err(_) => return SessionError::ConnectTimeout(CONNECT_TIMEOUT), }; let mut connection = Connection { stream, buf: BytesMut::with_capacity(64 * 1024), }; info!(%cache, "Connected to RPKI cache"); - self.status(|status| { - status.state = CacheState::Synchronising; - status.protocol_version = self.protocol_version; - }); + self.status(|status| status.state = CacheState::Synchronising); loop { if let Err(e) = self.synchronise(cache, &mut connection).await { @@ -527,7 +539,7 @@ impl Client { () = sleep_until(ask) => {} () = sleep_until(expires), if expires < ask => { self.expired(); - return SessionError::Timeout; + return SessionError::Expired; } } } @@ -592,10 +604,13 @@ impl Client { // one a minute would keep the old data in use for ever. let deadline = Instant::now() + ANSWER_TIMEOUT; let mut next = async || { - let wait = RESPONSE_TIMEOUT.min(deadline.saturating_duration_since(Instant::now())); - match timeout(wait, connection.next()).await { + let left = deadline.saturating_duration_since(Instant::now()); + match timeout(RESPONSE_TIMEOUT.min(left), connection.next()).await { Ok(pdu) => pdu, - Err(_) => Err(SessionError::Timeout), + Err(_) if left <= RESPONSE_TIMEOUT => { + Err(SessionError::AnswerTimeout(ANSWER_TIMEOUT)) + } + Err(_) => Err(SessionError::ResponseTimeout(RESPONSE_TIMEOUT)), } }; @@ -630,6 +645,10 @@ impl Client { _ => return Err(SessionError::Unexpected("data before a Cache Response")), } }; + // The cache has answered in the version asked for: that is the + // version of the session. + let protocol_version = self.protocol_version; + self.status(|status| status.protocol_version = protocol_version); if since.is_some_and(|place| place.session_id != session_id) { // The cache has started over; its serials mean something else // now. @@ -711,7 +730,8 @@ impl Client { serial, }); self.timers = timers.unwrap_or_default(); - self.expires = Some(Instant::now() + self.expire()); + let expires_in = self.expire(); + self.expires = Some(Instant::now() + expires_in); let changes = built.changed.vrps.len() + built.changed.aspa_customers.len(); if first || changes > 0 { self.publish(built.vrps, (!first).then_some(built.changed)); @@ -724,13 +744,13 @@ impl Client { changes, "RPKI data is up to date" ); - let expires_at = SystemTime::now() + self.expire(); + let updated_at = SystemTime::now(); self.status(|status| { status.state = CacheState::Synchronised; status.session_id = Some(session_id); status.serial = Some(serial); - status.updated_at = Some(SystemTime::now()); - status.expires_at = Some(expires_at); + status.updated_at = Some(updated_at); + status.expires_at = Some(updated_at + expires_in); status.last_error = None; }); Ok(Answer::Complete) diff --git a/crates/rpki/src/lib.rs b/crates/rpki/src/lib.rs index aaf1c33..96b351a 100644 --- a/crates/rpki/src/lib.rs +++ b/crates/rpki/src/lib.rs @@ -8,7 +8,6 @@ //! tell a route whose origin is authorised from one whose origin is not: //! the commonest kind of hijack and of mistake. //! -//! //! The same way, a network can sign a statement of who its providers are //! (an ASPA). With those a router can tell a path that could have come //! about by networks passing routes on as they are paid to from one that diff --git a/crates/rpki/tests/client.rs b/crates/rpki/tests/client.rs index 21f5062..ff09da7 100644 --- a/crates/rpki/tests/client.rs +++ b/crates/rpki/tests/client.rs @@ -425,7 +425,10 @@ async fn aspas_are_taken_replaced_and_withdrawn() { PathValidity::Valid ); let changed = update.changed.expect("a change to what there was"); - assert_eq!(*changed.aspa_customers, [3333, 2121, 1101].into()); + assert_eq!( + *changed.aspa_customers, + [3333, 2121, 1101].into_iter().collect() + ); assert!(changed.vrps.is_empty(), "no VRP came or went"); assert_eq!(update.vrps.len(), 1); @@ -536,7 +539,7 @@ async fn a_cache_that_speaks_version_one_has_no_aspas() { assert!(update.aspas.is_empty()); assert_eq!(update.vrps.len(), 1); let changed = update.changed.expect("the difference to what there was"); - assert_eq!(*changed.aspa_customers, [3333].into()); + assert_eq!(*changed.aspa_customers, [3333].into_iter().collect()); assert!(changed.vrps.is_empty()); let status = rpki.status(); assert_eq!((status.protocol_version, status.aspas), (1, 0));