use std::collections::BTreeMap;
use std::hash::Hash;
use std::net::Ipv4Addr;
use ntk_common::{Cost, HCoord, Naddr};
use ntk_netlink::{
Interface, Ipv4Net, Netlink, Nexthop, RouteKey, RouteSpec, RouteTarget, RuleSelector, RuleSpec,
TableAllocator, TableAllocatorError,
};
use ntk_qspn::{ArcId, RoutePath, RouteSnapshot};
use crate::kernel::addressing::{self, AddressingError};
const IDENTITY_ADDRESS_INTERFACE: &str = "lo";
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct AppliedDelta {
pub added: usize,
pub changed: usize,
pub removed: usize,
}
#[derive(Debug)]
pub struct RouteInstaller<K> {
kernel: K,
my_naddr: Naddr,
table: u32,
rule_priority: u32,
arc_endpoints: BTreeMap<ArcId, (Ipv4Addr, Interface)>,
applied: BTreeMap<HCoord, RouteSpec>,
identity_address: Option<Ipv4Net>,
rule_installed: bool,
}
impl<K> RouteInstaller<K> {
pub fn is_installed(&self) -> bool {
self.identity_address.is_some() || self.rule_installed || !self.applied.is_empty()
}
#[cfg(any(test, feature = "test-util"))]
pub fn kernel_ref(&self) -> &K {
&self.kernel
}
}
impl<K: Netlink> RouteInstaller<K> {
pub fn new(kernel: K, my_naddr: Naddr, table: u32, rule_priority: u32) -> Self {
Self {
kernel,
my_naddr,
table,
rule_priority,
arc_endpoints: BTreeMap::new(),
applied: BTreeMap::new(),
identity_address: None,
rule_installed: false,
}
}
pub fn realize(&mut self, real_naddr: Naddr) {
self.my_naddr = real_naddr;
}
pub fn set_arc_endpoint(&mut self, arc: ArcId, via: Ipv4Addr, dev: Interface) {
self.arc_endpoints.insert(arc, (via, dev));
}
pub fn clear_arc_endpoint(&mut self, arc: ArcId) {
self.arc_endpoints.remove(&arc);
}
pub async fn install_identity(&mut self) -> Result<(), RouteError> {
if self.my_naddr.is_virtual() {
return Ok(());
}
let address = addressing::host_address(&self.my_naddr)?;
match self
.kernel
.add_address(&Interface::name(IDENTITY_ADDRESS_INTERFACE), address)
.await
{
Ok(()) => {}
Err(error) if error.is_already_exists() => {}
Err(error) => return Err(error.into()),
}
self.identity_address = Some(address);
match self
.kernel
.add_rule(&RuleSpec {
table: self.table,
priority: self.rule_priority,
selector: RuleSelector::Any,
})
.await
{
Ok(()) => {}
Err(error) if error.is_already_exists() => {}
Err(error) => return Err(error.into()),
}
self.rule_installed = true;
Ok(())
}
pub async fn apply(&mut self, snapshot: &RouteSnapshot) -> Result<AppliedDelta, RouteError> {
if self.my_naddr.is_virtual() {
return Ok(AppliedDelta::default());
}
let src = self.identity_address.map(|net| net.address());
let mut next = BTreeMap::new();
for entry in snapshot.levels.iter().flatten() {
let destination = match addressing::gnode_destination(&self.my_naddr, entry.destination)
{
Ok(destination) => destination,
Err(error) => {
tracing::warn!(
level = entry.destination.level,
pos = entry.destination.pos,
%error,
"route snapshot entry dropped: destination cannot be encoded"
);
continue;
}
};
let target = route_target(&entry.paths, &self.arc_endpoints, src);
next.insert(
entry.destination,
RouteSpec {
destination,
table: self.table,
target,
},
);
}
let mut delta = AppliedDelta::default();
let mut first_error: Option<ntk_netlink::NetlinkError> = None;
for (destination, spec) in &next {
let is_new = self
.applied
.get(destination)
.is_none_or(|previous| previous != spec);
if !is_new {
continue;
}
let result = if self.applied.contains_key(destination) {
self.kernel.change_route(spec).await
} else {
self.kernel.add_route(spec).await
};
match result {
Ok(()) => {
if self.applied.contains_key(destination) {
delta.changed += 1;
} else {
delta.added += 1;
}
self.applied.insert(*destination, spec.clone());
}
Err(error) => {
tracing::warn!(%error, ?destination, "route install failed; will retry");
first_error.get_or_insert(error);
}
}
}
let stale: Vec<HCoord> = self
.applied
.keys()
.filter(|destination| !next.contains_key(destination))
.copied()
.collect();
for destination in stale {
let previous = self
.applied
.get(&destination)
.expect("destination came from self.applied's own keys")
.clone();
match self
.kernel
.remove_route(RouteKey {
destination: previous.destination,
table: previous.table,
})
.await
{
Ok(()) => {}
Err(error) if error.is_not_found() => {}
Err(error) => {
tracing::warn!(%error, ?destination, "stale route removal failed; will retry");
first_error.get_or_insert(error);
continue;
}
}
self.applied.remove(&destination);
delta.removed += 1;
}
match first_error {
Some(error) => Err(error.into()),
None => Ok(delta),
}
}
pub async fn teardown(&mut self) -> Result<(), RouteError> {
let mut first_error: Option<ntk_netlink::NetlinkError> = None;
for (destination, spec) in std::mem::take(&mut self.applied) {
match self
.kernel
.remove_route(RouteKey {
destination: spec.destination,
table: spec.table,
})
.await
{
Ok(()) => {}
Err(error) if error.is_not_found() => {}
Err(error) => {
first_error.get_or_insert(error);
self.applied.insert(destination, spec);
}
}
}
if self.rule_installed {
let rule = RuleSpec {
table: self.table,
priority: self.rule_priority,
selector: RuleSelector::Any,
};
match self.kernel.remove_rule(&rule).await {
Ok(()) => self.rule_installed = false,
Err(error) if error.is_not_found() => self.rule_installed = false,
Err(error) => {
first_error.get_or_insert(error);
}
}
}
if let Some(address) = self.identity_address {
match self
.kernel
.remove_address(&Interface::name(IDENTITY_ADDRESS_INTERFACE), address)
.await
{
Ok(()) => self.identity_address = None,
Err(error) if error.is_not_found() => self.identity_address = None,
Err(error) => {
first_error.get_or_insert(error);
}
}
}
match first_error {
Some(error) => Err(error.into()),
None => Ok(()),
}
}
}
struct Usable {
via: Ipv4Addr,
dev: Interface,
cost: Cost,
}
fn resolve_usable(
paths: &[RoutePath],
endpoints: &BTreeMap<ArcId, (Ipv4Addr, Interface)>,
) -> Vec<Usable> {
paths
.iter()
.filter(|path| path.cost != Cost::Dead)
.filter_map(|path| {
let (via, dev) = endpoints.get(&path.arc)?;
Some(Usable {
via: *via,
dev: dev.clone(),
cost: path.cost,
})
})
.collect()
}
fn route_target(
paths: &[RoutePath],
endpoints: &BTreeMap<ArcId, (Ipv4Addr, Interface)>,
src: Option<Ipv4Addr>,
) -> RouteTarget {
let usable = dedupe_by_gateway(resolve_usable(paths, endpoints));
match usable.as_slice() {
[] => RouteTarget::Unreachable,
[only] => RouteTarget::Gateway {
via: only.via,
dev: only.dev.clone(),
src,
},
several => {
let best = several
.iter()
.map(|u| effective_cost(u.cost))
.min()
.expect("non-empty slice");
RouteTarget::Multipath(
several
.iter()
.map(|u| Nexthop {
via: u.via,
dev: u.dev.clone(),
weight: nexthop_weight(u.cost, best),
})
.collect(),
)
}
}
}
fn dedupe_by_gateway(usable: Vec<Usable>) -> Vec<Usable> {
let mut seen = std::collections::HashSet::new();
usable
.into_iter()
.filter(|u| seen.insert((u.via, u.dev.clone())))
.collect()
}
fn effective_cost(cost: Cost) -> u64 {
match cost {
Cost::Null => 1,
Cost::Finite(magnitude) => magnitude.max(1),
Cost::Dead => unreachable!("Dead-cost paths are filtered out by resolve_usable"),
}
}
fn nexthop_weight(cost: Cost, best: u64) -> u8 {
let effective = effective_cost(cost);
let real_weight = (256u128 * u128::from(best) / u128::from(effective)).clamp(1, 256);
(real_weight - 1) as u8
}
#[derive(Debug, thiserror::Error)]
pub enum RouteError {
#[error(transparent)]
Netlink(#[from] ntk_netlink::NetlinkError),
#[error(transparent)]
Addressing(#[from] AddressingError),
}
#[derive(Debug, thiserror::Error)]
pub enum GenerationTableError {
#[error("route installer still has kernel state installed; call teardown() first")]
StillInstalled,
#[error(transparent)]
Allocator(#[from] TableAllocatorError),
}
pub fn release_generation_table<K, O>(
installer: &RouteInstaller<K>,
table_allocator: &mut TableAllocator<O>,
owner: &O,
) -> Result<u32, GenerationTableError>
where
O: Clone + Eq + Hash,
{
if installer.is_installed() {
return Err(GenerationTableError::StillInstalled);
}
Ok(table_allocator.release(owner)?)
}
#[cfg(test)]
mod tests {
use super::*;
use ntk_qspn::RouteEntry;
use std::sync::Arc;
#[test]
fn nexthop_weight_gives_cheapest_path_max_weight() {
let best = effective_cost(Cost::Finite(10));
assert_eq!(nexthop_weight(Cost::Finite(10), best), 255);
assert_eq!(nexthop_weight(Cost::Finite(20), best), 127);
assert_eq!(nexthop_weight(Cost::Finite(1000), best), 1);
}
#[test]
fn nexthop_weight_treats_null_cost_as_best() {
let best = effective_cost(Cost::Null);
assert_eq!(nexthop_weight(Cost::Null, best), 255);
}
fn path(arc: u32, cost: u64) -> RoutePath {
RoutePath {
arc: ArcId::from(arc),
hops: Vec::new(),
cost: Cost::Finite(cost),
nodes_inside: 0,
}
}
#[test]
fn route_target_dedupes_nexthops_sharing_a_gateway() {
let via1: Ipv4Addr = "169.254.1.1".parse().unwrap();
let via2: Ipv4Addr = "169.254.2.2".parse().unwrap();
let dev1 = Interface::name("eth0");
let dev2 = Interface::name("eth1");
let mut endpoints = BTreeMap::new();
endpoints.insert(ArcId::from(1), (via1, dev1.clone()));
endpoints.insert(ArcId::from(2), (via2, dev2.clone()));
let paths = vec![path(1, 10), path(1, 15), path(2, 20)];
let target = route_target(&paths, &endpoints, None);
let RouteTarget::Multipath(nexthops) = target else {
panic!("expected a multipath target, got {target:?}");
};
assert_eq!(
nexthops.len(),
2,
"the two arc-1 paths must collapse into a single nexthop: {nexthops:?}"
);
let best = effective_cost(Cost::Finite(10));
assert_eq!(
nexthops,
vec![
Nexthop {
via: via1,
dev: dev1,
weight: nexthop_weight(Cost::Finite(10), best),
},
Nexthop {
via: via2,
dev: dev2,
weight: nexthop_weight(Cost::Finite(20), best),
},
]
);
}
#[test]
fn route_target_collapses_single_gateway_to_plain_gateway() {
let via: Ipv4Addr = "169.254.9.9".parse().unwrap();
let dev = Interface::name("eth0");
let mut endpoints = BTreeMap::new();
endpoints.insert(ArcId::from(1), (via, dev.clone()));
let paths = vec![path(1, 10), path(1, 25)];
let target = route_target(&paths, &endpoints, None);
assert_eq!(
target,
RouteTarget::Gateway {
via,
dev,
src: None,
}
);
}
#[tokio::test]
async fn virtual_naddr_suppresses_both_identity_and_route_installation() {
let topology = ntk_common::Topology::new([4, 2]).unwrap();
let virtual_naddr = Naddr::new_allowing_virtual(topology.clone(), [4, 0]).unwrap();
assert!(virtual_naddr.is_virtual());
let kernel = ntk_netlink::FakeNetlink::with_links(vec![ntk_netlink::LinkInfo {
index: 1,
name: "lo".into(),
is_up: true,
}]);
let mut installer = RouteInstaller::new(kernel, virtual_naddr.clone(), 200, 9_990);
installer
.install_identity()
.await
.expect("a virtual naddr must not error, just skip");
assert!(
installer.kernel_ref().operations().is_empty(),
"no address/rule must ever be installed for a virtual position"
);
let snapshot = RouteSnapshot {
levels: vec![
vec![RouteEntry {
destination: HCoord::new(0, 1),
paths: vec![RoutePath {
arc: ArcId::from(1),
hops: Vec::new(),
cost: Cost::Finite(10),
nodes_inside: 1,
}],
}],
Vec::new(),
],
};
let delta = installer
.apply(&snapshot)
.await
.expect("a virtual naddr must not error, just skip");
assert_eq!(
delta,
AppliedDelta::default(),
"no route diff must ever be computed against a virtual position"
);
assert!(
installer.kernel_ref().operations().is_empty(),
"apply must not touch the kernel while the position is virtual, even with real \
destinations in the snapshot"
);
let real_naddr = Naddr::new(topology, [1, 0]).unwrap();
let mut real_installer = RouteInstaller::new(
ntk_netlink::FakeNetlink::with_links(vec![ntk_netlink::LinkInfo {
index: 1,
name: "lo".into(),
is_up: true,
}]),
real_naddr,
200,
9_990,
);
real_installer
.install_identity()
.await
.expect("a real naddr installs normally");
assert!(!real_installer.kernel_ref().operations().is_empty());
}
#[tokio::test]
async fn realize_transitions_a_virtual_installer_to_real_in_place() {
let topology = ntk_common::Topology::new([4, 2]).unwrap();
let virtual_naddr = Naddr::new_allowing_virtual(topology.clone(), [4, 0]).unwrap();
let kernel = ntk_netlink::FakeNetlink::with_links(vec![
ntk_netlink::LinkInfo {
index: 1,
name: "lo".into(),
is_up: true,
},
ntk_netlink::LinkInfo {
index: 2,
name: "eth0".into(),
is_up: true,
},
]);
let mut installer = RouteInstaller::new(kernel, virtual_naddr, 200, 9_990);
let via: Ipv4Addr = "169.254.1.1".parse().unwrap();
installer.set_arc_endpoint(ArcId::from(1), via, Interface::name("eth0"));
installer
.install_identity()
.await
.expect("still virtual: must not error");
assert!(
installer.kernel_ref().operations().is_empty(),
"still virtual: no kernel writes yet"
);
let real_naddr = Naddr::new(topology, [1, 0]).unwrap();
installer.realize(real_naddr);
installer
.install_identity()
.await
.expect("now real: installs for the first time");
assert!(
!installer.kernel_ref().operations().is_empty(),
"realize must let install_identity actually write kernel state"
);
let snapshot = RouteSnapshot {
levels: vec![
vec![RouteEntry {
destination: HCoord::new(0, 1),
paths: vec![RoutePath {
arc: ArcId::from(1),
hops: Vec::new(),
cost: Cost::Finite(10),
nodes_inside: 1,
}],
}],
Vec::new(),
],
};
let delta = installer.apply(&snapshot).await.unwrap();
assert_eq!(
delta.added, 1,
"the arc endpoint recorded while virtual must resolve now that the position is real"
);
}
#[tokio::test]
async fn apply_skips_unencodable_entry_and_withdraws_its_previous_route() {
let topology = ntk_common::Topology::new([4, 2, 2, 2]).unwrap();
let naddr = Naddr::new(topology, [1, 0, 1, 0]).unwrap();
let kernel = ntk_netlink::FakeNetlink::with_links(vec![ntk_netlink::LinkInfo {
index: 1,
name: "lo".into(),
is_up: true,
}]);
let mut installer = RouteInstaller::new(kernel, naddr, 200, 9_990);
let stable = HCoord::new(1, 1); let stale = HCoord::new(2, 0);
let baseline = RouteSnapshot {
levels: vec![
Vec::new(),
vec![RouteEntry {
destination: stable,
paths: vec![path(1, 10)],
}],
vec![RouteEntry {
destination: stale,
paths: vec![path(2, 10)],
}],
Vec::new(),
],
};
let baseline_delta = installer.apply(&baseline).await.unwrap();
assert_eq!(baseline_delta.added, 2, "both valid destinations install");
installer.kernel_ref().clear_operations();
let fresh = HCoord::new(0, 3); let corrupt = HCoord::new(2, 7); let next_snapshot = RouteSnapshot {
levels: vec![
vec![RouteEntry {
destination: fresh,
paths: vec![path(3, 10)],
}],
vec![RouteEntry {
destination: stable,
paths: vec![path(1, 10)],
}],
vec![RouteEntry {
destination: corrupt,
paths: vec![path(4, 10)],
}],
Vec::new(),
],
};
let delta = installer
.apply(&next_snapshot)
.await
.expect("one unencodable entry must not abort the whole snapshot");
assert_eq!(delta.added, 1, "fresh is added");
assert_eq!(delta.changed, 0, "stable is unchanged, no op issued for it");
assert_eq!(
delta.removed, 1,
"stale's previously-installed route is withdrawn: a corrupt update can no longer \
vouch for a CIDR it can no longer even name, so the safer state is unrouted, not a \
route nothing currently corroborates"
);
let ops = installer.kernel_ref().operations();
assert_eq!(
ops.len(),
2,
"exactly the good destinations' ops, never zero and never the bad one: {ops:?}"
);
assert!(
matches!(&ops[0], ntk_netlink::Operation::AddRoute(spec) if spec.table == 200),
"fresh must still be added: {ops:?}"
);
assert!(
matches!(&ops[1], ntk_netlink::Operation::RemoveRoute(key) if key.table == 200),
"stale's stale route must be withdrawn: {ops:?}"
);
}
#[tokio::test]
async fn two_installers_on_different_tables_never_collide() {
let topology = ntk_common::Topology::new([4, 2]).unwrap();
let main_naddr = Naddr::new(topology.clone(), [1, 0]).unwrap();
let bridge_naddr = Naddr::new(topology, [2, 0]).unwrap();
let kernel = Arc::new(ntk_netlink::FakeNetlink::with_links(vec![
ntk_netlink::LinkInfo {
index: 1,
name: "lo".into(),
is_up: true,
},
ntk_netlink::LinkInfo {
index: 2,
name: "eth0".into(),
is_up: true,
},
]));
let mut allocator: TableAllocator<&str> = TableAllocator::new();
let (bridge_table, bridge_priority) = allocator.acquire("bridge").unwrap();
assert_ne!(bridge_table, allocator.main_table());
assert_ne!(bridge_priority, allocator.main_rule_priority());
let mut main_installer = RouteInstaller::new(
crate::node::kernel_handle::KernelHandle(kernel.clone()),
main_naddr,
allocator.main_table(),
allocator.main_rule_priority(),
);
let mut bridge_installer = RouteInstaller::new(
crate::node::kernel_handle::KernelHandle(kernel.clone()),
bridge_naddr,
bridge_table,
bridge_priority,
);
main_installer.install_identity().await.unwrap();
bridge_installer.install_identity().await.unwrap();
let via: Ipv4Addr = "169.254.1.1".parse().unwrap();
main_installer.set_arc_endpoint(ArcId::from(1), via, Interface::name("eth0"));
bridge_installer.set_arc_endpoint(ArcId::from(2), via, Interface::name("eth0"));
let snapshot = |arc: u32| RouteSnapshot {
levels: vec![
vec![RouteEntry {
destination: HCoord::new(0, 3),
paths: vec![path(arc, 10)],
}],
Vec::new(),
],
};
main_installer.apply(&snapshot(1)).await.unwrap();
bridge_installer.apply(&snapshot(2)).await.unwrap();
let ops = kernel.operations();
let rules: Vec<_> = ops
.iter()
.filter_map(|op| match op {
ntk_netlink::Operation::AddRule(r) => Some(*r),
_ => None,
})
.collect();
assert_eq!(rules.len(), 2, "one AddRule per identity: {ops:?}");
assert_ne!(rules[0].table, rules[1].table);
assert_ne!(rules[0].priority, rules[1].priority);
let routes: Vec<_> = ops
.iter()
.filter_map(|op| match op {
ntk_netlink::Operation::AddRoute(r) => Some(r.clone()),
_ => None,
})
.collect();
assert_eq!(
routes.len(),
2,
"one AddRoute per identity, same destination: {ops:?}"
);
assert_ne!(
routes[0].table, routes[1].table,
"identical destination in two different tables must not collide: {ops:?}"
);
}
#[tokio::test]
async fn release_before_teardown_is_rejected_then_the_table_is_reusable() {
let topology = ntk_common::Topology::new([4, 2]).unwrap();
let naddr = Naddr::new(topology, [1, 0]).unwrap();
let kernel = ntk_netlink::FakeNetlink::with_links(vec![ntk_netlink::LinkInfo {
index: 1,
name: "lo".into(),
is_up: true,
}]);
let mut allocator: TableAllocator<&str> = TableAllocator::new();
let (table, priority) = allocator.acquire("bridge").unwrap();
let mut installer = RouteInstaller::new(kernel, naddr, table, priority);
installer.install_identity().await.unwrap();
assert!(matches!(
release_generation_table(&installer, &mut allocator, &"bridge"),
Err(GenerationTableError::StillInstalled)
));
assert_eq!(
allocator.table_of(&"bridge"),
Some((table, priority)),
"a rejected release must not touch the allocator's bookkeeping"
);
installer.teardown().await.unwrap();
let released = release_generation_table(&installer, &mut allocator, &"bridge").unwrap();
assert_eq!(released, table);
let (reused, _) = allocator.acquire("another-bridge").unwrap();
assert_eq!(
reused, table,
"a released table must be the next one handed out"
);
}
}
#[cfg(test)]
mod resilience_tests {
use super::*;
use ntk_netlink::{FakeNetlink, LinkInfo, RouteTable, RuleTable};
use ntk_qspn::RouteEntry;
pub(super) fn links() -> Vec<LinkInfo> {
vec![LinkInfo {
index: 1,
name: "lo".into(),
is_up: true,
}]
}
pub(super) fn path(arc: u32) -> RoutePath {
RoutePath {
arc: ArcId::from(arc),
hops: Vec::new(),
cost: Cost::Finite(10),
nodes_inside: 0,
}
}
pub(super) fn snapshot(destinations: &[(HCoord, u32)]) -> RouteSnapshot {
RouteSnapshot {
levels: vec![
destinations
.iter()
.map(|(destination, arc)| RouteEntry {
destination: *destination,
paths: vec![path(*arc)],
})
.collect(),
Vec::new(),
],
}
}
pub(super) fn installer() -> RouteInstaller<FakeNetlink> {
let topology = ntk_common::Topology::new([4, 2]).unwrap();
let naddr = Naddr::new(topology, [1, 0]).unwrap();
let mut installer =
RouteInstaller::new(FakeNetlink::with_links(links()), naddr, 200, 9_990);
let via: Ipv4Addr = "169.254.1.1".parse().unwrap();
for arc in 1..=3 {
installer.set_arc_endpoint(ArcId::from(arc), via, Interface::name("lo"));
}
installer
}
#[tokio::test]
async fn route_already_flushed_by_the_kernel_does_not_wedge_apply_or_teardown() {
let mut installer = installer();
installer.install_identity().await.unwrap();
let (a, b) = (HCoord::new(0, 2), HCoord::new(0, 3));
installer.apply(&snapshot(&[(a, 1), (b, 2)])).await.unwrap();
for spec in installer.kernel_ref().list_routes(Some(200)).await.unwrap() {
installer
.kernel_ref()
.remove_route(RouteKey {
destination: spec.destination,
table: 200,
})
.await
.unwrap();
}
let delta = installer.apply(&snapshot(&[])).await.unwrap();
assert_eq!(delta.removed, 2);
assert!(
installer.applied.is_empty(),
"vanished routes must be forgotten"
);
installer.apply(&snapshot(&[(a, 1)])).await.unwrap();
installer
.kernel_ref()
.remove_route(RouteKey {
destination: installer.applied[&a].destination,
table: 200,
})
.await
.unwrap();
installer.teardown().await.unwrap();
assert!(!installer.is_installed());
}
#[tokio::test]
async fn install_identity_adopts_leftover_address_and_rule_from_an_interrupted_run() {
let mut installer = installer();
installer.install_identity().await.unwrap();
installer.install_identity().await.unwrap();
installer.teardown().await.unwrap();
assert!(!installer.is_installed());
assert!(
installer
.kernel_ref()
.list_rules()
.await
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn teardown_removes_rule_and_address_even_when_a_route_removal_fails() {
let mut installer = installer();
installer.install_identity().await.unwrap();
let dest = HCoord::new(0, 2);
installer.apply(&snapshot(&[(dest, 1)])).await.unwrap();
let prefix = installer.applied[&dest].destination;
installer.kernel_ref().arm_route_failure(
prefix,
ntk_netlink::NetlinkError::PermissionDenied("denied".into()),
);
assert!(installer.teardown().await.is_err());
assert!(
installer
.kernel_ref()
.list_rules()
.await
.unwrap()
.is_empty(),
"the rule must still be removed after a route failure"
);
assert!(installer.identity_address.is_none());
assert!(installer.is_installed(), "the failed route stays tracked");
installer.teardown().await.unwrap();
assert!(!installer.is_installed());
}
#[tokio::test]
async fn one_failing_destination_does_not_starve_the_rest_of_the_batch() {
let mut installer = installer();
installer.install_identity().await.unwrap();
let (a, b, c) = (HCoord::new(0, 1), HCoord::new(0, 2), HCoord::new(0, 3));
installer.apply(&snapshot(&[(c, 3)])).await.unwrap();
let broken = addressing::gnode_destination(&installer.my_naddr, a).unwrap();
installer.kernel_ref().arm_route_failure(
broken,
ntk_netlink::NetlinkError::PermissionDenied("denied".into()),
);
let result = installer.apply(&snapshot(&[(a, 1), (b, 2)])).await;
assert!(result.is_err(), "the batch error must be reported");
assert!(
installer.applied.contains_key(&b),
"later destination installed"
);
assert!(!installer.applied.contains_key(&c), "stale loop still ran");
assert!(!installer.applied.contains_key(&a));
let delta = installer.apply(&snapshot(&[(a, 1), (b, 2)])).await.unwrap();
assert_eq!(delta.added, 1, "only the failed destination is retried");
}
}