Kernel and RPKI: say what went wrong, and once

The RPKI client had one Timeout error whose text said "did not answer
within 60s" whatever had run out: the connect timeout, the time for a
whole answer, or the data's lifetime. These are what /api/rpki shows as
the last error; each is its own now. The protocol version shown is the
one the cache answered in, not the one being tried. The VRP set, hashed
by the hundred thousand with every answer, uses the quick hash the ASPA
table already did.

In the kernel installer the writer task's error was worked out in three
places, and the next-hop resolution checked for our own networks twice
on every cache miss. A refusal was logged twice, once with the status
lock held. The interfaces are listed in order of index by the structure
that holds them, from a function that can be tested on any OS, and is.
EIO is with the other errnos; Plan has no installed_prefixes (no
callers) and clears its records in one place, with the order of
Unresolved::ALL asserted at compile time since indexing depends on it.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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<u32, Link>,
+    /// By index, which is the order they are given out in too.
+    links: BTreeMap<u32, Link>,
     addresses: HashSet<Address>,
 }
 
@@ -64,11 +66,31 @@ impl Interfaces {
         }
     }
 
-    /// The interfaces, by index.
+    /// The interfaces, in order of index.
     pub fn links(&self) -> impl Iterator<Item = &Link> {
         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<InterfaceStatus> {
+        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<u32> = 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());
+    }
 }

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<Resolving, Result<Target, Unresolved>, foldhash::fast::RandomState>,
+    resolved: FastHashMap<Resolving, Result<Target, Unresolved>>,
     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<HashMap<Target, u32, foldhash::fast::RandomState>>,
+    nexthops: Option<FastHashMap<Target, u32>>,
     /// 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<InterfaceStatus> = 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<Result<(), FibError>, 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?;

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<R>(&self, change: impl FnOnce(&mut Status) -> R) -> R {
         change(
             &mut self
                 .shared
                 .status
                 .lock()
                 .unwrap_or_else(PoisonError::into_inner),
-        );
+        )
     }
 }
 

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::<u8>(), 0..100),
         ) {
             let wire = message(message_type, 1, &body, &[]);

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<K, V> = std::collections::HashMap<K, V, foldhash::fast::RandomState>;
-type HashSet<K> = std::collections::HashSet<K, foldhash::fast::RandomState>;
+pub(crate) type HashMap<K, V> = std::collections::HashMap<K, V, foldhash::fast::RandomState>;
+pub(crate) type HashSet<K> = std::collections::HashSet<K, foldhash::fast::RandomState>;
 
 /// 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<Item = &IpPrefix> {
-        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<Op> {
-        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 })

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<K> = std::collections::HashSet<K, foldhash::fast::RandomState>;
+
 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)

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

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