use ntkd::kernel::addressing;
use ntkd::kernel::config::NtkdConfig;
use ntkd::node::adapters::QspnArcResolverAdapter;
use ntkd::node::dispatch::Dispatcher;
use ntkd::node::lifecycle::{self, Dialer, NodeInputs, StartedNode};
use ntkd::node::peers::PeerLinks;
use ntkd::node::registry::{LinkRegistry, encode_caller_id};
use ntkd::node::stubs::IdentityStubFactoryAdapter;
use std::collections::HashMap;
use std::net::Ipv4Addr;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use anyhow::Context;
use futures::future::BoxFuture;
use ntk_common::{Cost, HCoord, Naddr, Topology};
use ntk_neighborhood::{
Arc as NeighborArc, FakeIpRouteManager, FixedRttProbe, LocalNic, NeighborhoodConfig,
NeighborhoodRpcHandler, NeighborhoodStubFactory, NeighborhoodTiming, NodeId,
};
use ntk_netlink::{
AddressTable, FakeNetlink, Interface, Ipv4Net, LinkInfo, Operation, RouteKey, RouteTable,
RouteTarget, RuleSelector, RuleSpec, RuleTable,
};
use ntk_proto::v1::{CallerContext, MethodCall, RemoteError, ResponsePayload, TypedValue};
use ntk_rpc::{FakeRpcClient, RpcClient, RpcHandler};
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
fn topology() -> Topology {
Topology::new([4, 2, 2, 2]).unwrap()
}
fn config() -> NtkdConfig {
NtkdConfig::from_str("gsizes = [4, 2, 2, 2]\nnics = []\nport = 269\n").unwrap()
}
fn position(idx: u32) -> Vec<u32> {
vec![idx, 0, 0, 0]
}
fn naddr(idx: u32) -> Naddr {
Naddr::new(topology(), position(idx)).unwrap()
}
fn nic_address(idx: u32, n: u8) -> Ipv4Addr {
Ipv4Addr::new(169, 254, idx as u8 + 1, n)
}
fn nic_mac(idx: u32, domain_index: usize) -> String {
format!("02:00:00:{idx:02x}:{domain_index:02x}:00")
}
#[derive(Debug, Default)]
struct Medium {
domains: Mutex<HashMap<String, Vec<(String, ntk_neighborhood::Handle)>>>,
dispatchers: Mutex<HashMap<u8, Arc<Dispatcher>>>,
}
impl Medium {
fn join_domain(&self, domain: &str, node_id: &str, handle: ntk_neighborhood::Handle) {
self.domains
.lock()
.unwrap()
.entry(domain.to_owned())
.or_default()
.push((node_id.to_owned(), handle));
}
fn peers_on(&self, domain: &str, self_id: &str) -> Vec<ntk_neighborhood::Handle> {
self.domains
.lock()
.unwrap()
.get(domain)
.into_iter()
.flatten()
.filter(|(id, _)| id != self_id)
.map(|(_, h)| h.clone())
.collect()
}
fn register_dispatcher(&self, idx: u32, dispatcher: Arc<Dispatcher>) {
self.dispatchers
.lock()
.unwrap()
.insert(idx as u8, dispatcher);
}
fn dispatcher_for_addr(&self, addr: &str) -> Option<Arc<Dispatcher>> {
let ip: Ipv4Addr = addr.parse().ok()?;
let idx = ip.octets()[2].wrapping_sub(1);
self.dispatchers.lock().unwrap().get(&idx).cloned()
}
}
#[derive(Debug)]
struct FanOutHandler(Vec<NeighborhoodRpcHandler>);
impl RpcHandler for FanOutHandler {
fn handle<'a>(
&'a self,
caller: CallerContext,
unicast_id: TypedValue,
call: MethodCall,
auth: Option<ntk_proto::v1::Auth>,
) -> BoxFuture<'a, Result<ResponsePayload, RemoteError>> {
Box::pin(async move {
for target in self.0.clone() {
let caller = caller.clone();
let unicast_id = unicast_id.clone();
let call = call.clone();
let auth = auth.clone();
tokio::spawn(async move {
let _ = target.handle(caller, unicast_id, call, auth).await;
});
}
Ok(ResponsePayload::default())
})
}
}
#[derive(Debug)]
struct TestNeighborhoodStubFactory {
self_id: String,
medium: Arc<Medium>,
}
impl NeighborhoodStubFactory for TestNeighborhoodStubFactory {
fn broadcast(&self, dev: &str) -> Arc<dyn RpcClient> {
let targets = self
.medium
.peers_on(dev, &self.self_id)
.into_iter()
.map(|h| NeighborhoodRpcHandler::for_broadcast(h, dev.to_owned()))
.collect();
Arc::new(FakeRpcClient::new(Arc::new(FanOutHandler(targets))))
}
fn unicast(&self, arc: &NeighborArc) -> Arc<dyn RpcClient> {
let peer = self
.medium
.peers_on(&arc.my_dev, &self.self_id)
.into_iter()
.next()
.unwrap_or_else(|| {
panic!(
"test medium: no peer on domain {} for {}",
arc.my_dev, self.self_id
)
});
Arc::new(FakeRpcClient::new(Arc::new(
NeighborhoodRpcHandler::for_unicast(peer),
)))
}
}
#[derive(Debug)]
struct TestDialer {
medium: Arc<Medium>,
}
impl Dialer for TestDialer {
fn dial(&self, addr: &str, _port: u16) -> BoxFuture<'_, Option<Arc<dyn RpcClient>>> {
let target = self.medium.dispatcher_for_addr(addr);
Box::pin(
async move { target.map(|d| Arc::new(FakeRpcClient::new(d)) as Arc<dyn RpcClient>) },
)
}
}
fn addr_allocator(idx: u32) -> Box<dyn FnMut() -> String + Send> {
let mut n = 0u8;
Box::new(move || {
n += 1;
nic_address(idx, n).to_string()
})
}
struct SimNode {
tasks: JoinSet<()>,
cancel: CancellationToken,
kernel: Arc<FakeNetlink>,
neighborhood: ntk_neighborhood::Handle,
started: StartedNode<FakeNetlink>,
}
impl SimNode {
fn qspn(&self) -> ntk_qspn::QspnHandle {
self.started.running.generation.borrow().qspn.clone()
}
fn route_table(&self) -> u32 {
self.started.running.route_table
}
}
async fn spawn_node(idx: u32, domains: &[&str], medium: &Arc<Medium>) -> SimNode {
let mut links = vec![LinkInfo {
index: 1,
name: "lo".into(),
is_up: true,
}];
for (i, dev) in domains.iter().enumerate() {
links.push(LinkInfo {
index: (i + 2) as u32,
name: (*dev).to_owned(),
is_up: true,
});
}
let neighborhood_kernel = FakeNetlink::with_links(links.clone());
let routing_kernel = Arc::new(FakeNetlink::with_links(links));
let self_id = format!("node-{idx}");
let stub_factory = Arc::new(TestNeighborhoodStubFactory {
self_id: self_id.clone(),
medium: medium.clone(),
});
let my_id = NodeId::from_raw((idx + 1) as i32).unwrap();
let neighborhood_config = NeighborhoodConfig {
my_id,
max_arcs: 8,
kernel: neighborhood_kernel,
stub_factory,
ip_route_manager: Arc::new(FakeIpRouteManager::new()),
rtt_probe: Arc::new(FixedRttProbe(Some(10))),
timing: NeighborhoodTiming {
radar_interval: Duration::from_millis(15),
arc_monitor_interval: (Duration::from_millis(1), Duration::from_millis(3)),
},
new_linklocal_address: addr_allocator(idx),
signing_key: None,
require_auth: false,
};
let mut tasks = JoinSet::new();
let cancel = CancellationToken::new();
let (neighborhood, neighborhood_join) =
ntk_neighborhood::Manager::spawn(neighborhood_config, cancel.child_token());
tasks.spawn(async move {
let _ = neighborhood_join.await;
});
for dev in domains {
medium.join_domain(dev, &self_id, neighborhood.clone());
}
let dialer = Arc::new(TestDialer {
medium: medium.clone(),
});
let started = lifecycle::run(
NodeInputs {
config: config(),
neighborhood: neighborhood.clone(),
registry: Arc::new(LinkRegistry::new()),
links: Arc::new(PeerLinks::new()),
routing_kernel: routing_kernel.clone(),
dialer,
initial_position: Some(position(idx)),
preformed: None,
my_id,
},
&mut tasks,
cancel.child_token(),
)
.await
.expect("lifecycle::run");
tokio::time::sleep(Duration::from_millis(5)).await;
for (i, dev) in domains.iter().enumerate() {
neighborhood
.start_monitor(LocalNic {
dev: (*dev).to_owned(),
mac: nic_mac(idx, i),
})
.await
.expect("start_monitor");
}
medium.register_dispatcher(idx, started.dispatcher.clone());
SimNode {
tasks,
cancel,
kernel: routing_kernel,
neighborhood,
started,
}
}
async fn wait_until(mut check: impl FnMut() -> bool, timeout: Duration) -> bool {
let deadline = tokio::time::Instant::now() + timeout;
loop {
if check() {
return true;
}
if tokio::time::Instant::now() >= deadline {
return false;
}
tokio::time::sleep(Duration::from_millis(2)).await;
}
}
fn cost_at(snapshot: &ntk_qspn::RouteSnapshot, pos: u32) -> Option<Cost> {
snapshot
.levels
.first()?
.iter()
.find(|e| e.destination == HCoord::new(0, pos))?
.paths
.first()
.map(|p| p.cost)
}
fn route_destination(op: &Operation) -> Option<&Ipv4Net> {
match op {
Operation::AddRoute(spec) | Operation::ChangeRoute(spec) => Some(&spec.destination),
Operation::RemoveRoute(key) => Some(&key.destination),
_ => None,
}
}
const CAPABILITY_PROBE_TABLE: u32 = 0xFFFF_FFF0;
fn is_capability_probe(op: &Operation) -> bool {
match op {
Operation::AddRoute(spec) | Operation::ChangeRoute(spec) => {
spec.table == CAPABILITY_PROBE_TABLE
}
Operation::RemoveRoute(key) => key.table == CAPABILITY_PROBE_TABLE,
_ => false,
}
}
fn expected_route(my_idx: u32, dest_pos: u32, via: Ipv4Addr, dev: &str) -> ntk_netlink::RouteSpec {
ntk_netlink::RouteSpec {
destination: addressing::gnode_destination(&naddr(my_idx), HCoord::new(0, dest_pos))
.unwrap(),
table: ntk_netlink::DEFAULT_MAIN_TABLE_ID,
target: RouteTarget::Gateway {
via,
dev: Interface::name(dev),
src: Some(addressing::host_address(&naddr(my_idx)).unwrap().address()),
},
}
}
const CONVERGE_TIMEOUT: Duration = Duration::from_secs(5);
#[test]
fn inbound_caller_never_resolves_to_a_different_peers_arc_on_link_id_collision() {
let receiver_id = NodeId::from_raw(1).unwrap();
let peer_a_id = NodeId::from_raw(101).unwrap();
let peer_b_id = NodeId::from_raw(202).unwrap();
let peer_a_registry = LinkRegistry::new();
let peer_a_link = peer_a_registry.link_for_neighbour(receiver_id, "receiver-mac@a", "eth0");
let peer_b_registry = LinkRegistry::new();
let peer_b_link = peer_b_registry.link_for_neighbour(receiver_id, "receiver-mac@b", "eth0");
assert_eq!(
peer_a_link, peer_b_link,
"peer A and peer B must independently mint the same first LinkId for this test to \
exercise the real collision, not a hypothetical one"
);
let receiver_registry = Arc::new(LinkRegistry::new());
let receiver_link_to_a = receiver_registry.link_for_neighbour(peer_a_id, "a-mac", "eth0");
let receiver_link_to_b = receiver_registry.link_for_neighbour(peer_b_id, "b-mac", "eth1");
let qspn_arc_a = ntk_qspn::ArcId::from(11u32);
let qspn_arc_b = ntk_qspn::ArcId::from(22u32);
receiver_registry.set_qspn_arc(receiver_link_to_a, qspn_arc_a);
receiver_registry.set_qspn_arc(receiver_link_to_b, qspn_arc_b);
let caller_from_a = CallerContext {
source_id: None,
src_nic: Some(encode_caller_id(peer_a_id)),
};
let caller_from_b = CallerContext {
source_id: None,
src_nic: Some(encode_caller_id(peer_b_id)),
};
let qspn_resolver = QspnArcResolverAdapter {
registry: receiver_registry.clone(),
};
assert_eq!(
ntk_qspn::ArcResolver::resolve(&qspn_resolver, &caller_from_a),
Some(qspn_arc_a),
"peer A's inbound call must resolve to the receiver's own arc-to-A"
);
assert_eq!(
ntk_qspn::ArcResolver::resolve(&qspn_resolver, &caller_from_b),
Some(qspn_arc_b),
"peer B's inbound call must resolve to the receiver's own arc-to-B -- never arc-to-A --\
despite peer A and peer B having minted the identical LinkId(1) locally"
);
let identity_stub_factory = IdentityStubFactoryAdapter {
links: Arc::new(PeerLinks::new()),
registry: receiver_registry.clone(),
};
assert_eq!(
ntk_identities::IdentityStubFactory::arc_for_caller(&identity_stub_factory, &caller_from_a),
Some(receiver_link_to_a.identities()),
"peer A's inbound call must resolve to the receiver's own identities arc-to-A"
);
assert_eq!(
ntk_identities::IdentityStubFactory::arc_for_caller(&identity_stub_factory, &caller_from_b),
Some(receiver_link_to_b.identities()),
"peer B's inbound call must resolve to the receiver's own identities arc-to-B -- never \
arc-to-A"
);
}
#[tokio::test]
async fn chain_converges_then_arc_flap_reinstalls_only_the_affected_route() {
let medium = Arc::new(Medium::default());
let node0 = spawn_node(0, &["link-a"], &medium).await;
let node1 = spawn_node(1, &["link-a", "link-b"], &medium).await;
let node2 = spawn_node(2, &["link-b"], &medium).await;
let converged = wait_until(
|| {
cost_at(&node0.qspn().snapshot(), 1) == Some(Cost::Finite(10))
&& cost_at(&node0.qspn().snapshot(), 2) == Some(Cost::Finite(20))
&& cost_at(&node1.qspn().snapshot(), 0) == Some(Cost::Finite(10))
&& cost_at(&node1.qspn().snapshot(), 2) == Some(Cost::Finite(10))
&& cost_at(&node2.qspn().snapshot(), 1) == Some(Cost::Finite(10))
&& cost_at(&node2.qspn().snapshot(), 0) == Some(Cost::Finite(20))
},
CONVERGE_TIMEOUT,
)
.await;
if !converged {
eprintln!("node0 snapshot: {:?}", node0.qspn().snapshot());
eprintln!("node1 snapshot: {:?}", node1.qspn().snapshot());
eprintln!("node2 snapshot: {:?}", node2.qspn().snapshot());
eprintln!(
"node0 arcs: {:?}",
node0.neighborhood.snapshot().borrow().clone()
);
eprintln!(
"node1 arcs: {:?}",
node1.neighborhood.snapshot().borrow().clone()
);
eprintln!(
"node2 arcs: {:?}",
node2.neighborhood.snapshot().borrow().clone()
);
eprintln!(
"node1 current_arcs: {:?}",
node1.qspn().current_arcs().await
);
}
assert!(
converged,
"chain did not converge to the expected route set in time"
);
let expected0_1 = expected_route(0, 1, nic_address(1, 1), "link-a");
let expected0_2 = expected_route(0, 2, nic_address(1, 1), "link-a");
let expected1_0 = expected_route(1, 0, nic_address(0, 1), "link-a");
let expected1_2 = expected_route(1, 2, nic_address(2, 1), "link-b");
let expected2_1 = expected_route(2, 1, nic_address(1, 2), "link-b");
let expected2_0 = expected_route(2, 0, nic_address(1, 2), "link-b");
let installed = wait_until(
|| {
node0
.kernel
.operations()
.iter()
.any(|op| op == &Operation::AddRoute(expected0_1.clone()))
&& node0
.kernel
.operations()
.iter()
.any(|op| op == &Operation::AddRoute(expected0_2.clone()))
&& node1
.kernel
.operations()
.iter()
.any(|op| op == &Operation::AddRoute(expected1_0.clone()))
&& node1
.kernel
.operations()
.iter()
.any(|op| op == &Operation::AddRoute(expected1_2.clone()))
&& node2
.kernel
.operations()
.iter()
.any(|op| op == &Operation::AddRoute(expected2_1.clone()))
&& node2
.kernel
.operations()
.iter()
.any(|op| op == &Operation::AddRoute(expected2_0.clone()))
},
CONVERGE_TIMEOUT,
)
.await;
assert!(
installed,
"converged routes were not installed into the kernel model"
);
let node1_routes = node1
.kernel
.list_routes(Some(node1.route_table()))
.await
.unwrap();
assert_eq!(
node1_routes.len(),
2,
"node1 must have exactly the two converged destinations installed, nothing stale"
);
node1.kernel.clear_operations();
node1.neighborhood.stop_monitor("link-a").await.unwrap();
node1
.neighborhood
.start_monitor(LocalNic {
dev: "link-a".to_owned(),
mac: nic_mac(1, 0),
})
.await
.unwrap();
let reconverged = wait_until(
|| cost_at(&node1.qspn().snapshot(), 0) == Some(Cost::Finite(10)),
CONVERGE_TIMEOUT,
)
.await;
assert!(
reconverged,
"arc to node0 did not reconverge after the flap"
);
let reinstalled = wait_until(
|| {
node1
.kernel
.operations()
.iter()
.any(|op| op == &Operation::AddRoute(expected1_0.clone()))
},
CONVERGE_TIMEOUT,
)
.await;
assert!(
reinstalled,
"route to node0 was not reinstalled, identically, after the flap"
);
let ops_after_flap = node1.kernel.operations();
assert!(
ops_after_flap.iter().any(
|op| matches!(op, Operation::RemoveRoute(k) if k.destination == expected1_0.destination)
),
"flap must remove the stale route to node0 before reinstalling it: {ops_after_flap:?}"
);
assert!(
!ops_after_flap
.iter()
.any(|op| route_destination(op) == Some(&expected1_2.destination)),
"flap of the link to node0 must not touch node1's unrelated route to node2: {ops_after_flap:?}"
);
}
#[tokio::test]
async fn cancellation_reaps_tasks_and_teardown_is_exact_inverse_of_install() {
let medium = Arc::new(Medium::default());
let mut node0 = spawn_node(0, &["link-a"], &medium).await;
let _node1 = spawn_node(1, &["link-a"], &medium).await;
let converged = wait_until(
|| cost_at(&node0.qspn().snapshot(), 1) == Some(Cost::Finite(10)),
CONVERGE_TIMEOUT,
)
.await;
assert!(converged, "arc did not establish in time");
let expected_route0_1 = expected_route(0, 1, nic_address(1, 1), "link-a");
let installed = wait_until(
|| {
node0
.kernel
.operations()
.iter()
.any(|op| op == &Operation::AddRoute(expected_route0_1.clone()))
},
CONVERGE_TIMEOUT,
)
.await;
assert!(installed, "route to node1 was not installed in time");
let raw_install_ops = node0.kernel.operations();
let install_ops: Vec<Operation> = raw_install_ops
.iter()
.filter(|op| !is_capability_probe(op))
.cloned()
.collect();
let expected_install = vec![
Operation::AddAddress {
interface: Interface::name("lo"),
network: addressing::host_address(&naddr(0)).unwrap(),
},
Operation::AddRule(RuleSpec {
table: ntk_netlink::DEFAULT_MAIN_TABLE_ID,
priority: ntk_netlink::DEFAULT_MAIN_RULE_PRIORITY,
selector: RuleSelector::Any,
}),
Operation::AddRoute(expected_route0_1.clone()),
];
assert_eq!(
install_ops, expected_install,
"install log must be exactly [address, rule, route] with no churn"
);
node0.cancel.cancel();
let tasks = &mut node0.tasks;
let drained = tokio::time::timeout(Duration::from_secs(5), async {
while tasks.join_next().await.is_some() {}
})
.await;
assert!(
drained.is_ok(),
"JoinSet did not drain within 5s after cancelling the root token — some task ignored it"
);
node0
.started
.running
.route_installer
.lock()
.await
.teardown()
.await
.unwrap();
let after_teardown = node0.kernel.operations();
let teardown_ops = &after_teardown[raw_install_ops.len()..];
let expected_teardown = vec![
Operation::RemoveRoute(RouteKey {
destination: expected_route0_1.destination,
table: ntk_netlink::DEFAULT_MAIN_TABLE_ID,
}),
Operation::RemoveRule(RuleSpec {
table: ntk_netlink::DEFAULT_MAIN_TABLE_ID,
priority: ntk_netlink::DEFAULT_MAIN_RULE_PRIORITY,
selector: RuleSelector::Any,
}),
Operation::RemoveAddress {
interface: Interface::name("lo"),
network: addressing::host_address(&naddr(0)).unwrap(),
},
];
assert_eq!(
teardown_ops,
expected_teardown.as_slice(),
"teardown log must be the exact reverse of the install log"
);
assert!(
node0
.kernel
.list_routes(Some(node0.route_table()))
.await
.unwrap()
.is_empty()
);
assert!(node0.kernel.list_rules().await.unwrap().is_empty());
assert!(node0.kernel.list_addresses(None).await.unwrap().is_empty());
}
#[derive(Debug)]
struct NamespaceReport {
label: &'static str,
arcs: Vec<NeighborArc>,
expected_destination: Ipv4Net,
route_installed: bool,
routes: Vec<ntk_netlink::RouteSpec>,
addresses: Vec<ntk_netlink::AddressEntry>,
}
struct NamespaceSpec {
label: &'static str,
my_id: NodeId,
my_idx: u32,
peer_idx: u32,
dev: &'static str,
port: u16,
fd_tx: std::sync::mpsc::Sender<std::os::fd::RawFd>,
moved_rx: std::sync::mpsc::Receiver<()>,
my_done_tx: Option<tokio::sync::oneshot::Sender<()>>,
peer_done_rx: Option<tokio::sync::oneshot::Receiver<()>>,
report_tx: tokio::sync::oneshot::Sender<anyhow::Result<NamespaceReport>>,
}
async fn link_index(handle: &rtnetlink::Handle, name: &str) -> anyhow::Result<u32> {
use futures::TryStreamExt;
handle
.link()
.get()
.match_name(name.to_owned())
.execute()
.try_next()
.await
.with_context(|| format!("resolving link {name:?}"))?
.map(|link| link.header.index)
.ok_or_else(|| anyhow::anyhow!("link {name:?} not found"))
}
async fn bring_link_up(handle: &rtnetlink::Handle, name: &str) -> anyhow::Result<()> {
let index = link_index(handle, name).await?;
handle
.link()
.change(rtnetlink::LinkUnspec::new_with_index(index).up().build())
.execute()
.await
.with_context(|| format!("bringing {name:?} up"))
}
async fn spawn_real_node(
my_id: NodeId,
position: Vec<u32>,
dev: &str,
port: u16,
tasks: &mut JoinSet<()>,
cancel: CancellationToken,
) -> anyhow::Result<StartedNode<ntk_netlink::RealNetlink>> {
use ntk_rpc::{TcpServer, UdpBroadcaster};
use ntkd::node::ip_route::RealIpRouteManager;
use ntkd::node::lifecycle::{TcpDialer, linklocal_allocator, synthetic_mac};
use ntkd::node::peers::PeerLinks;
use ntkd::node::registry::LinkRegistry;
use ntkd::node::stubs::NeighborhoodStubFactoryAdapter;
let config = NtkdConfig::from_str(&format!(
"gsizes = [4, 2, 2, 2]\nnics = [\"{dev}\"]\nport = {port}\n"
))?;
let registry = Arc::new(LinkRegistry::new());
let links = Arc::new(PeerLinks::new());
let broadcaster = Arc::new(UdpBroadcaster::bind(Some(dev), port, 1 << 16)?);
let mut broadcasters = HashMap::new();
broadcasters.insert(dev.to_owned(), broadcaster);
let neighborhood_config = NeighborhoodConfig {
my_id,
max_arcs: 64,
kernel: ntk_netlink::RealNetlink::new()?,
stub_factory: Arc::new(NeighborhoodStubFactoryAdapter {
broadcasters: broadcasters.clone(),
links: links.clone(),
registry: registry.clone(),
}),
ip_route_manager: Arc::new(RealIpRouteManager {
kernel: ntk_netlink::RealNetlink::new()?,
}),
rtt_probe: Arc::new(FixedRttProbe(Some(0))),
timing: NeighborhoodTiming {
radar_interval: Duration::from_millis(200),
arc_monitor_interval: (Duration::from_millis(20), Duration::from_millis(40)),
},
new_linklocal_address: linklocal_allocator(my_id),
signing_key: None,
require_auth: false,
};
let (neighborhood, neighborhood_join) =
ntk_neighborhood::Manager::spawn(neighborhood_config, cancel.child_token());
tasks.spawn(async move {
let _ = neighborhood_join.await;
});
neighborhood
.start_monitor(LocalNic {
dev: dev.to_owned(),
mac: synthetic_mac(dev, my_id),
})
.await?;
let routing_kernel = Arc::new(ntk_netlink::RealNetlink::new()?);
let started = lifecycle::run(
NodeInputs {
config,
neighborhood: neighborhood.clone(),
registry,
links,
routing_kernel,
dialer: Arc::new(TcpDialer::default()),
initial_position: Some(position),
preformed: None,
my_id,
},
tasks,
cancel.clone(),
)
.await?;
let server = TcpServer::bind(format!("0.0.0.0:{port}").parse()?, 1 << 20).await?;
let dispatcher = started.dispatcher.clone();
let server_cancel = cancel.child_token();
tasks.spawn(async move {
server.serve(dispatcher, server_cancel).await;
});
for (dev, broadcaster) in broadcasters {
let handler = Arc::new(NeighborhoodRpcHandler::for_broadcast(
neighborhood.clone(),
dev,
));
let broadcast_cancel = cancel.child_token();
tasks.spawn(async move {
ntk_neighborhood::serve_broadcast(broadcaster, handler, broadcast_cancel).await;
});
}
Ok(started)
}
async fn namespace_body(spec: &mut NamespaceSpec) -> anyhow::Result<NamespaceReport> {
let (connection, handle, _) = rtnetlink::new_connection()
.with_context(|| format!("{}: rtnetlink connection", spec.label))?;
tokio::spawn(connection);
bring_link_up(&handle, "lo")
.await
.with_context(|| format!("{}: bring lo up", spec.label))?;
bring_link_up(&handle, spec.dev)
.await
.with_context(|| format!("{}: bring {} up", spec.label, spec.dev))?;
drop(handle);
let mut tasks = JoinSet::new();
let cancel = CancellationToken::new();
let started = spawn_real_node(
spec.my_id,
position(spec.my_idx),
spec.dev,
spec.port,
&mut tasks,
cancel.clone(),
)
.await
.with_context(|| format!("{}: compose real node", spec.label))?;
let observer = ntk_netlink::RealNetlink::new()
.with_context(|| format!("{}: observer RealNetlink", spec.label))?;
let expected_destination =
addressing::gnode_destination(&naddr(spec.my_idx), HCoord::new(0, spec.peer_idx))
.with_context(|| format!("{}: expected destination", spec.label))?;
let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
let (route_installed, routes, arcs) = loop {
let arcs: Vec<NeighborArc> = started.running.neighborhood.snapshot().borrow().clone();
let routes = observer
.list_routes(Some(started.running.route_table))
.await
.with_context(|| format!("{}: list_routes", spec.label))?;
let found = routes.iter().any(|r| r.destination == expected_destination);
if found || tokio::time::Instant::now() >= deadline {
break (found, routes, arcs);
}
tokio::time::sleep(Duration::from_millis(200)).await;
};
let addresses = observer
.list_addresses(None)
.await
.with_context(|| format!("{}: list_addresses", spec.label))?;
let my_done_tx = spec
.my_done_tx
.take()
.expect("my_done_tx is taken exactly once, here");
let peer_done_rx = spec
.peer_done_rx
.take()
.expect("peer_done_rx is taken exactly once, here");
let _ = my_done_tx.send(());
let _ = peer_done_rx.await;
cancel.cancel();
while tasks.join_next().await.is_some() {}
if let Err(err) = started
.running
.route_installer
.lock()
.await
.teardown()
.await
{
tracing::warn!(%err, "{}: route teardown failed", spec.label);
}
Ok(NamespaceReport {
label: spec.label,
arcs,
expected_destination,
route_installed,
routes,
addresses,
})
}
fn run_namespace_worker(mut spec: NamespaceSpec) {
let outcome = (|| -> anyhow::Result<NamespaceReport> {
use std::os::fd::AsRawFd;
nix::sched::unshare(nix::sched::CloneFlags::CLONE_NEWNET)
.map_err(|errno| anyhow::anyhow!("{}: unshare(CLONE_NEWNET): {errno}", spec.label))?;
let ns_file = std::fs::File::open("/proc/thread-self/ns/net")
.with_context(|| format!("{}: open own netns fd", spec.label))?;
spec.fd_tx
.send(ns_file.as_raw_fd())
.map_err(|_| anyhow::anyhow!("{}: coordinator dropped fd channel", spec.label))?;
spec.moved_rx.recv().map_err(|_| {
anyhow::anyhow!("{}: coordinator dropped veth-moved signal", spec.label)
})?;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.with_context(|| format!("{}: build current-thread runtime", spec.label))?;
let report = rt.block_on(namespace_body(&mut spec));
drop(ns_file);
report
})();
let _ = spec.report_tx.send(outcome);
}
#[ignore = "requires the equivalent of CAP_NET_ADMIN over its own network namespaces"]
#[tokio::test]
async fn real_netns_two_daemons_establish_arc_and_route() {
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("debug"));
let _ = tracing_subscriber::fmt()
.with_env_filter(filter)
.with_test_writer()
.try_init();
const VETH_A: &str = "ntkd-test-a";
const VETH_B: &str = "ntkd-test-b";
const PORT: u16 = 27269;
let (connection, root, _) = rtnetlink::new_connection().expect("rtnetlink connection");
tokio::spawn(connection);
root.link()
.add(rtnetlink::LinkVeth::new(VETH_A, VETH_B).build())
.execute()
.await
.expect("create veth pair");
let idx_a = link_index(&root, VETH_A)
.await
.expect("resolve ntkd-test-a");
let idx_b = link_index(&root, VETH_B)
.await
.expect("resolve ntkd-test-b");
let (fd_tx_a, fd_rx_a) = std::sync::mpsc::channel();
let (fd_tx_b, fd_rx_b) = std::sync::mpsc::channel();
let (moved_tx_a, moved_rx_a) = std::sync::mpsc::channel();
let (moved_tx_b, moved_rx_b) = std::sync::mpsc::channel();
let (report_tx_a, report_rx_a) = tokio::sync::oneshot::channel();
let (report_tx_b, report_rx_b) = tokio::sync::oneshot::channel();
let (done_tx_a, done_rx_a) = tokio::sync::oneshot::channel();
let (done_tx_b, done_rx_b) = tokio::sync::oneshot::channel();
let thread_a = std::thread::spawn(move || {
run_namespace_worker(NamespaceSpec {
label: "ns-a",
my_id: NodeId::from_raw(101).unwrap(),
my_idx: 0,
peer_idx: 1,
dev: VETH_A,
port: PORT,
fd_tx: fd_tx_a,
moved_rx: moved_rx_a,
my_done_tx: Some(done_tx_a),
peer_done_rx: Some(done_rx_b),
report_tx: report_tx_a,
});
});
let thread_b = std::thread::spawn(move || {
run_namespace_worker(NamespaceSpec {
label: "ns-b",
my_id: NodeId::from_raw(102).unwrap(),
my_idx: 1,
peer_idx: 0,
dev: VETH_B,
port: PORT,
fd_tx: fd_tx_b,
moved_rx: moved_rx_b,
my_done_tx: Some(done_tx_b),
peer_done_rx: Some(done_rx_a),
report_tx: report_tx_b,
});
});
let fd_a = fd_rx_a.recv().expect("ns-a netns fd");
let fd_b = fd_rx_b.recv().expect("ns-b netns fd");
root.link()
.change(
rtnetlink::LinkUnspec::new_with_index(idx_a)
.setns_by_fd(fd_a)
.build(),
)
.execute()
.await
.expect("move ntkd-test-a into ns-a");
root.link()
.change(
rtnetlink::LinkUnspec::new_with_index(idx_b)
.setns_by_fd(fd_b)
.build(),
)
.execute()
.await
.expect("move ntkd-test-b into ns-b");
moved_tx_a.send(()).expect("signal ns-a");
moved_tx_b.send(()).expect("signal ns-b");
let report_a = report_rx_a.await.expect("ns-a report channel");
let report_b = report_rx_b.await.expect("ns-b report channel");
thread_a
.join()
.unwrap_or_else(|e| panic!("ns-a worker thread panicked: {e:?}"));
thread_b
.join()
.unwrap_or_else(|e| panic!("ns-b worker thread panicked: {e:?}"));
let report_a = report_a.expect("ns-a namespace body");
let report_b = report_b.expect("ns-b namespace body");
eprintln!(
"{}: arcs={:#?} routes={:#?}",
report_a.label, report_a.arcs, report_a.routes
);
eprintln!(
"{}: arcs={:#?} routes={:#?}",
report_b.label, report_b.arcs, report_b.routes
);
assert!(
report_a.arcs.iter().any(|a| a.cost.is_some()),
"ns-a never measured a cost for its neighbor — no arc established: {:#?}",
report_a.arcs
);
assert!(
report_b.arcs.iter().any(|a| a.cost.is_some()),
"ns-b never measured a cost for its neighbor — no arc established: {:#?}",
report_b.arcs
);
assert!(
report_a.route_installed,
"ns-a's real kernel routing table never gained a route to {} (its neighbor's g-node); \
routes: {:#?}, addresses: {:#?}, arcs: {:#?}",
report_a.expected_destination, report_a.routes, report_a.addresses, report_a.arcs
);
assert!(
report_b.route_installed,
"ns-b's real kernel routing table never gained a route to {} (its neighbor's g-node); \
routes: {:#?}, addresses: {:#?}, arcs: {:#?}",
report_b.expected_destination, report_b.routes, report_b.addresses, report_b.arcs
);
}
fn negotiation_topology() -> Topology {
Topology::new([8]).unwrap()
}
#[derive(Debug)]
struct NegotiationNamespaceReport {
label: &'static str,
initial_position: Vec<u32>,
final_position: Vec<u32>,
rehooked: bool,
route_count: usize,
routes: Vec<ntk_netlink::RouteSpec>,
addresses: Vec<ntk_netlink::AddressEntry>,
}
struct NegotiationNamespaceSpec {
label: &'static str,
my_id: NodeId,
dev: &'static str,
port: u16,
fd_tx: std::sync::mpsc::Sender<std::os::fd::RawFd>,
moved_rx: std::sync::mpsc::Receiver<()>,
my_done_tx: Option<tokio::sync::oneshot::Sender<()>>,
peer_done_rx: Option<tokio::sync::oneshot::Receiver<()>>,
report_tx: tokio::sync::oneshot::Sender<anyhow::Result<NegotiationNamespaceReport>>,
}
async fn spawn_real_negotiating_node(
my_id: NodeId,
dev: &str,
port: u16,
tasks: &mut JoinSet<()>,
cancel: CancellationToken,
) -> anyhow::Result<StartedNode<ntk_netlink::RealNetlink>> {
use ntk_rpc::{TcpServer, UdpBroadcaster};
use ntkd::node::ip_route::RealIpRouteManager;
use ntkd::node::lifecycle::{TcpDialer, linklocal_allocator, synthetic_mac};
use ntkd::node::peers::PeerLinks;
use ntkd::node::registry::LinkRegistry;
use ntkd::node::stubs::NeighborhoodStubFactoryAdapter;
let config = NtkdConfig::from_str(&format!(
"gsizes = [8]\nnics = [\"{dev}\"]\nport = {port}\n"
))?;
let registry = Arc::new(LinkRegistry::new());
let links = Arc::new(PeerLinks::new());
let broadcaster = Arc::new(UdpBroadcaster::bind(Some(dev), port, 1 << 16)?);
let mut broadcasters = HashMap::new();
broadcasters.insert(dev.to_owned(), broadcaster);
let neighborhood_config = NeighborhoodConfig {
my_id,
max_arcs: 64,
kernel: ntk_netlink::RealNetlink::new()?,
stub_factory: Arc::new(NeighborhoodStubFactoryAdapter {
broadcasters: broadcasters.clone(),
links: links.clone(),
registry: registry.clone(),
}),
ip_route_manager: Arc::new(RealIpRouteManager {
kernel: ntk_netlink::RealNetlink::new()?,
}),
rtt_probe: Arc::new(FixedRttProbe(Some(0))),
timing: NeighborhoodTiming {
radar_interval: Duration::from_millis(200),
arc_monitor_interval: (Duration::from_millis(20), Duration::from_millis(40)),
},
new_linklocal_address: linklocal_allocator(my_id),
signing_key: None,
require_auth: false,
};
let (neighborhood, neighborhood_join) =
ntk_neighborhood::Manager::spawn(neighborhood_config, cancel.child_token());
tasks.spawn(async move {
let _ = neighborhood_join.await;
});
neighborhood
.start_monitor(LocalNic {
dev: dev.to_owned(),
mac: synthetic_mac(dev, my_id),
})
.await?;
let routing_kernel = Arc::new(ntk_netlink::RealNetlink::new()?);
let started = lifecycle::run(
NodeInputs {
config,
neighborhood: neighborhood.clone(),
registry,
links,
routing_kernel,
dialer: Arc::new(TcpDialer::default()),
initial_position: None,
preformed: None,
my_id,
},
tasks,
cancel.clone(),
)
.await?;
let server = TcpServer::bind(format!("0.0.0.0:{port}").parse()?, 1 << 20).await?;
let dispatcher = started.dispatcher.clone();
let server_cancel = cancel.child_token();
tasks.spawn(async move {
server.serve(dispatcher, server_cancel).await;
});
for (dev, broadcaster) in broadcasters {
let handler = Arc::new(NeighborhoodRpcHandler::for_broadcast(
neighborhood.clone(),
dev,
));
let broadcast_cancel = cancel.child_token();
tasks.spawn(async move {
ntk_neighborhood::serve_broadcast(broadcaster, handler, broadcast_cancel).await;
});
}
Ok(started)
}
async fn negotiation_namespace_body(
spec: &mut NegotiationNamespaceSpec,
) -> anyhow::Result<NegotiationNamespaceReport> {
let (connection, handle, _) = rtnetlink::new_connection()
.with_context(|| format!("{}: rtnetlink connection", spec.label))?;
tokio::spawn(connection);
bring_link_up(&handle, "lo")
.await
.with_context(|| format!("{}: bring lo up", spec.label))?;
bring_link_up(&handle, spec.dev)
.await
.with_context(|| format!("{}: bring {} up", spec.label, spec.dev))?;
drop(handle);
let mut tasks = JoinSet::new();
let cancel = CancellationToken::new();
let started =
spawn_real_negotiating_node(spec.my_id, spec.dev, spec.port, &mut tasks, cancel.clone())
.await
.with_context(|| format!("{}: compose real negotiating node", spec.label))?;
let initial_position = started
.running
.generation
.borrow()
.qspn
.my_naddr()
.positions()
.to_vec();
let observer = ntk_netlink::RealNetlink::new()
.with_context(|| format!("{}: observer RealNetlink", spec.label))?;
let deadline = tokio::time::Instant::now() + Duration::from_secs(25);
let (final_position, routes, rehooked) = loop {
let generation = started.running.generation.borrow().clone();
let position = generation.qspn.my_naddr().positions().to_vec();
let routes = observer
.list_routes(Some(started.running.route_table))
.await
.with_context(|| format!("{}: list_routes", spec.label))?;
let rehooked_and_reattached = generation.rehooked && !routes.is_empty();
if rehooked_and_reattached || tokio::time::Instant::now() >= deadline {
break (position, routes, generation.rehooked);
}
tokio::time::sleep(Duration::from_millis(200)).await;
};
let addresses = observer
.list_addresses(None)
.await
.with_context(|| format!("{}: list_addresses", spec.label))?;
let my_done_tx = spec
.my_done_tx
.take()
.expect("my_done_tx is taken exactly once, here");
let peer_done_rx = spec
.peer_done_rx
.take()
.expect("peer_done_rx is taken exactly once, here");
let _ = my_done_tx.send(());
let _ = peer_done_rx.await;
cancel.cancel();
while tasks.join_next().await.is_some() {}
if let Err(err) = started
.running
.route_installer
.lock()
.await
.teardown()
.await
{
tracing::warn!(%err, "{}: route teardown failed", spec.label);
}
Ok(NegotiationNamespaceReport {
label: spec.label,
initial_position,
final_position,
rehooked,
route_count: routes.len(),
routes,
addresses,
})
}
fn run_negotiation_namespace_worker(mut spec: NegotiationNamespaceSpec) {
let outcome = (|| -> anyhow::Result<NegotiationNamespaceReport> {
use std::os::fd::AsRawFd;
nix::sched::unshare(nix::sched::CloneFlags::CLONE_NEWNET)
.map_err(|errno| anyhow::anyhow!("{}: unshare(CLONE_NEWNET): {errno}", spec.label))?;
let ns_file = std::fs::File::open("/proc/thread-self/ns/net")
.with_context(|| format!("{}: open own netns fd", spec.label))?;
spec.fd_tx
.send(ns_file.as_raw_fd())
.map_err(|_| anyhow::anyhow!("{}: coordinator dropped fd channel", spec.label))?;
spec.moved_rx.recv().map_err(|_| {
anyhow::anyhow!("{}: coordinator dropped veth-moved signal", spec.label)
})?;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.with_context(|| format!("{}: build current-thread runtime", spec.label))?;
let report = rt.block_on(negotiation_namespace_body(&mut spec));
drop(ns_file);
report
})();
let _ = spec.report_tx.send(outcome);
}
#[ignore = "requires the equivalent of CAP_NET_ADMIN over its own network namespaces"]
#[tokio::test]
async fn real_netns_two_daemons_negotiate_a_shared_network() {
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("debug"));
let _ = tracing_subscriber::fmt()
.with_env_filter(filter)
.with_test_writer()
.with_thread_names(true)
.try_init();
const VETH_A: &str = "ntkd-negot-a";
const VETH_B: &str = "ntkd-negot-b";
const PORT: u16 = 27271;
let (connection, root, _) = rtnetlink::new_connection().expect("rtnetlink connection");
tokio::spawn(connection);
root.link()
.add(rtnetlink::LinkVeth::new(VETH_A, VETH_B).build())
.execute()
.await
.expect("create veth pair");
let idx_a = link_index(&root, VETH_A)
.await
.expect("resolve ntkd-negot-a");
let idx_b = link_index(&root, VETH_B)
.await
.expect("resolve ntkd-negot-b");
let (fd_tx_a, fd_rx_a) = std::sync::mpsc::channel();
let (fd_tx_b, fd_rx_b) = std::sync::mpsc::channel();
let (moved_tx_a, moved_rx_a) = std::sync::mpsc::channel();
let (moved_tx_b, moved_rx_b) = std::sync::mpsc::channel();
let (report_tx_a, report_rx_a) = tokio::sync::oneshot::channel();
let (report_tx_b, report_rx_b) = tokio::sync::oneshot::channel();
let (done_tx_a, done_rx_a) = tokio::sync::oneshot::channel();
let (done_tx_b, done_rx_b) = tokio::sync::oneshot::channel();
let thread_a = std::thread::Builder::new()
.name("ns-a".to_string())
.spawn(move || {
run_negotiation_namespace_worker(NegotiationNamespaceSpec {
label: "ns-a",
my_id: NodeId::from_raw(201).unwrap(),
dev: VETH_A,
port: PORT,
fd_tx: fd_tx_a,
moved_rx: moved_rx_a,
my_done_tx: Some(done_tx_a),
peer_done_rx: Some(done_rx_b),
report_tx: report_tx_a,
});
})
.expect("spawn ns-a worker thread");
let thread_b = std::thread::Builder::new()
.name("ns-b".to_string())
.spawn(move || {
run_negotiation_namespace_worker(NegotiationNamespaceSpec {
label: "ns-b",
my_id: NodeId::from_raw(202).unwrap(),
dev: VETH_B,
port: PORT,
fd_tx: fd_tx_b,
moved_rx: moved_rx_b,
my_done_tx: Some(done_tx_b),
peer_done_rx: Some(done_rx_a),
report_tx: report_tx_b,
});
})
.expect("spawn ns-b worker thread");
let fd_a = fd_rx_a.recv().expect("ns-a netns fd");
let fd_b = fd_rx_b.recv().expect("ns-b netns fd");
root.link()
.change(
rtnetlink::LinkUnspec::new_with_index(idx_a)
.setns_by_fd(fd_a)
.build(),
)
.execute()
.await
.expect("move ntkd-negot-a into ns-a");
root.link()
.change(
rtnetlink::LinkUnspec::new_with_index(idx_b)
.setns_by_fd(fd_b)
.build(),
)
.execute()
.await
.expect("move ntkd-negot-b into ns-b");
moved_tx_a.send(()).expect("signal ns-a");
moved_tx_b.send(()).expect("signal ns-b");
let report_a = report_rx_a.await.expect("ns-a report channel");
let report_b = report_rx_b.await.expect("ns-b report channel");
thread_a
.join()
.unwrap_or_else(|e| panic!("ns-a worker thread panicked: {e:?}"));
thread_b
.join()
.unwrap_or_else(|e| panic!("ns-b worker thread panicked: {e:?}"));
let report_a = report_a.expect("ns-a namespace body");
let report_b = report_b.expect("ns-b namespace body");
eprintln!("{}: {report_a:#?}", report_a.label);
eprintln!("{}: {report_b:#?}", report_b.label);
assert_ne!(
report_a.rehooked, report_b.rehooked,
"exactly one side should adopt the negotiated position, never both or neither"
);
let (winner, loser) = if report_a.rehooked {
(&report_b, &report_a)
} else {
(&report_a, &report_b)
};
assert_eq!(
loser.final_position.len(),
1,
"single-level topology: a resolved position always has exactly one entry"
);
assert_ne!(
loser.final_position, winner.initial_position,
"the loser must resolve to a *free* slot in the winner's network, not the winner's own \
already-occupied position"
);
let topology = negotiation_topology();
let old_address = addressing::host_address(
&Naddr::new(topology.clone(), loser.initial_position.clone()).unwrap(),
)
.expect("loser's own trivial address is always addressable");
let new_address =
addressing::host_address(&Naddr::new(topology, loser.final_position.clone()).unwrap())
.expect("loser's negotiated address is always addressable");
let netsukuku_addresses: Vec<_> = loser
.addresses
.iter()
.filter(|a| a.network.address().octets()[0] == 10)
.collect();
assert_eq!(
netsukuku_addresses.len(),
1,
"the loser's real kernel address table should carry exactly one Netsukuku-space \
address (its negotiated one, {new_address}) — a second entry would mean its \
torn-down trivial-generation address {old_address} leaked alongside it: {:#?}",
loser.addresses
);
assert_eq!(
netsukuku_addresses[0].network, new_address,
"the loser's sole Netsukuku-space address is not its negotiated one: {:#?}",
loser.addresses
);
assert!(
winner.route_count >= 1,
"the winner's real kernel routing table never gained a route to its (migrated) peer: \
{:#?}",
winner.routes
);
assert!(
loser.route_count >= 1,
"the loser's real kernel routing table never gained a route to its (winner) peer after \
rehook re-attached the arc: {:#?}",
loser.routes
);
}