use std::collections::HashMap;
use std::collections::hash_map::RandomState;
use std::hash::{BuildHasher, Hasher};
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::time::Duration;
use futures::future::BoxFuture;
use ntk_common::{Fingerprint, Naddr, Topology};
use ntk_hooking::HookingEvent;
use ntk_identities::{ArcInfo, MigrationId};
use ntk_netlink::{Interface, TableAllocator};
use ntk_qspn::QspnEvent;
use ntk_rpc::RpcClient;
use tokio::sync::{broadcast, watch};
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
use crate::kernel::config::NtkdConfig;
use crate::kernel::routes::RouteInstaller;
use crate::node::adapters::{CoordinatorClientAdapter, NetworkInfo, QspnArcResolverAdapter};
use crate::node::dispatch::{Dispatcher, IdentityStack};
use crate::node::kernel_handle::{KernelHandle, SendNetlink};
use crate::node::peers::PeerLinks;
use crate::node::registry::LinkRegistry;
use crate::node::services::{self, HookingProvenance};
use crate::node::stubs::{
HookingStubFactoryAdapter, IdentityStubFactoryAdapter, QspnStubFactoryAdapter,
};
pub trait Dialer: Send + Sync {
fn dial(&self, addr: &str, port: u16) -> BoxFuture<'_, Option<Arc<dyn RpcClient>>>;
fn dial_via(
&self,
addr: &str,
port: u16,
device: Option<&str>,
) -> BoxFuture<'_, Option<Arc<dyn RpcClient>>> {
let _ = device;
self.dial(addr, port)
}
}
#[derive(Debug)]
pub struct TcpDialer {
pub max_frame_length: usize,
pub call_timeout: Duration,
}
impl Default for TcpDialer {
fn default() -> Self {
Self {
max_frame_length: 1 << 20,
call_timeout: Duration::from_secs(10),
}
}
}
impl Dialer for TcpDialer {
fn dial(&self, addr: &str, port: u16) -> BoxFuture<'_, Option<Arc<dyn RpcClient>>> {
self.dial_via(addr, port, None)
}
fn dial_via(
&self,
addr: &str,
port: u16,
device: Option<&str>,
) -> BoxFuture<'_, Option<Arc<dyn RpcClient>>> {
let addr = addr.to_owned();
let device = device.map(str::to_owned);
Box::pin(async move {
let socket_addr: std::net::SocketAddr = format!("{addr}:{port}").parse().ok()?;
match ntk_rpc::TcpRpcClient::connect_via(
socket_addr,
device.as_deref(),
self.max_frame_length,
self.call_timeout,
)
.await
{
Ok(client) => Some(Arc::new(client) as Arc<dyn RpcClient>),
Err(error) => {
tracing::debug!(
%socket_addr, device = ?device, %error,
"ntkd: outbound dial attempt failed"
);
None
}
}
})
}
}
#[must_use]
pub fn linklocal_allocator(my_id: ntk_neighborhood::NodeId) -> Box<dyn FnMut() -> String + Send> {
let mut index: u32 = 0;
Box::new(move || {
let addr = derive_linklocal(my_id, index);
index += 1;
addr.to_string()
})
}
const LINKLOCAL_VALID_THIRDS: u32 = 254;
const LINKLOCAL_FOURTHS: u32 = 256;
const LINKLOCAL_SPACE: u32 = LINKLOCAL_VALID_THIRDS * LINKLOCAL_FOURTHS;
fn derive_linklocal(salt: ntk_neighborhood::NodeId, index: u32) -> Ipv4Addr {
use std::hash::{Hash, Hasher};
let mut hasher = std::collections::hash_map::DefaultHasher::new();
salt.hash(&mut hasher);
let start = (hasher.finish() % u64::from(LINKLOCAL_SPACE)) as u32;
let slot = (start + index) % LINKLOCAL_SPACE;
let third = (slot / LINKLOCAL_FOURTHS) as u8 + 1; let fourth = (slot % LINKLOCAL_FOURTHS) as u8; Ipv4Addr::new(169, 254, third, fourth)
}
fn derive_initial_position(my_id: ntk_neighborhood::NodeId, topology: &Topology) -> Vec<u32> {
use std::hash::{Hash, Hasher};
(0..topology.levels())
.map(|level| {
let gsize = topology.gsize(level).unwrap_or(1).max(1);
let mut hasher = std::collections::hash_map::DefaultHasher::new();
my_id.hash(&mut hasher);
level.hash(&mut hasher);
(hasher.finish() % u64::from(gsize)) as u32
})
.collect()
}
fn virtual_placeholder(topology: &Topology) -> Naddr {
let pos: Vec<u32> = (0..topology.levels())
.map(|level| topology.gsize(level).unwrap_or(1))
.collect();
Naddr::new_allowing_virtual(topology.clone(), pos)
.expect("gsize(level) at every level has exactly topology.levels() entries")
}
#[derive(Clone, Debug)]
pub struct PreformedNetwork {
pub network_id: i64,
pub position: Vec<u32>,
}
pub struct NodeInputs<K> {
pub config: NtkdConfig,
pub neighborhood: ntk_neighborhood::Handle,
pub registry: Arc<LinkRegistry>,
pub links: Arc<PeerLinks>,
pub routing_kernel: Arc<K>,
pub dialer: Arc<dyn Dialer>,
pub initial_position: Option<Vec<u32>>,
pub preformed: Option<PreformedNetwork>,
pub my_id: ntk_neighborhood::NodeId,
}
impl<K: std::fmt::Debug> std::fmt::Debug for NodeInputs<K> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("NodeInputs")
.field("config", &self.config)
.field("neighborhood", &self.neighborhood)
.field("registry", &self.registry)
.field("links", &self.links)
.field("routing_kernel", &self.routing_kernel)
.field("dialer", &"<dyn Dialer>")
.field("initial_position", &self.initial_position)
.field("preformed", &self.preformed)
.field("my_id", &self.my_id)
.finish()
}
}
#[derive(Clone, Debug)]
pub struct GenerationHandles {
pub qspn: ntk_qspn::QspnHandle,
pub peers: ntk_peerservices::Handle,
pub coordinator: ntk_coordinator::Handle,
pub andna: ntk_andna::Handle,
pub rehooked: bool,
pub migrations: u32,
}
impl GenerationHandles {
fn from_generation<K>(generation: &Generation<K>, migrations: u32) -> Self {
Self {
qspn: generation.qspn.clone(),
peers: generation.peers.clone(),
coordinator: generation.coordinator.clone(),
andna: generation.andna.clone(),
rehooked: migrations > 0,
migrations,
}
}
}
#[derive(Debug)]
pub struct RunningNode<K> {
pub generation: watch::Receiver<GenerationHandles>,
pub identities: ntk_identities::Handle,
pub neighborhood: ntk_neighborhood::Handle,
pub hooking: ntk_hooking::HookingHandle,
pub registry: Arc<LinkRegistry>,
pub route_table: u32,
pub route_installer: Arc<tokio::sync::Mutex<RouteInstaller<KernelHandle<K>>>>,
pub kernel: Arc<K>,
pub net: Arc<NetworkInfo>,
}
#[derive(Debug)]
pub struct StartedNode<K> {
pub running: RunningNode<K>,
pub dispatcher: Arc<Dispatcher>,
}
struct Generation<K> {
qspn: ntk_qspn::QspnHandle,
qspn_events: broadcast::Receiver<QspnEvent>,
hooking: ntk_hooking::HookingHandle,
peers: ntk_peerservices::Handle,
coordinator: ntk_coordinator::Handle,
andna: ntk_andna::Handle,
dispatch: IdentityStack,
route_installer: RouteInstaller<KernelHandle<K>>,
}
#[derive(Clone, Copy)]
enum QspnOrigin {
CreateNet,
Entering {
guest_gnode_level: usize,
host_gnode_level: usize,
},
}
#[allow(clippy::too_many_arguments)]
async fn bootstrap_generation<K>(
topology: Topology,
my_naddr: Naddr,
origin: QspnOrigin,
hooking: HookingProvenance,
net: Arc<NetworkInfo>,
registry: Arc<LinkRegistry>,
links: Arc<PeerLinks>,
my_id: ntk_neighborhood::NodeId,
kernel: KernelHandle<K>,
table: u32,
rule_priority: u32,
tasks: &mut JoinSet<()>,
cancel: CancellationToken,
signing_key: Option<ed25519_dalek::SigningKey>,
require_auth: bool,
coordinator_handoff: Option<ntk_coordinator::HandOff>,
) -> anyhow::Result<Generation<K>>
where
K: SendNetlink + 'static,
{
let levels = topology.levels();
let route_naddr = match origin {
QspnOrigin::CreateNet => my_naddr.clone(),
QspnOrigin::Entering { .. } => virtual_placeholder(&topology),
};
let my_fingerprint = Fingerprint::new(random_fingerprint_id(), 0, vec![0u32; levels]);
let qspn_stub_factory = Arc::new(QspnStubFactoryAdapter {
links: links.clone(),
registry: registry.clone(),
my_id,
});
let (qspn, qspn_join) = match origin {
QspnOrigin::CreateNet => ntk_qspn::spawn(
my_naddr.clone(),
my_fingerprint,
ntk_qspn::QspnConfig::default(),
qspn_stub_factory,
Arc::new(ntk_qspn::FixedThreshold::default()),
Arc::new(ntk_qspn::DefaultArcIdSource::default()),
cancel.child_token(),
),
QspnOrigin::Entering {
guest_gnode_level,
host_gnode_level,
} => ntk_qspn::spawn_entering(
my_naddr.clone(),
my_fingerprint,
ntk_qspn::QspnConfig::default(),
qspn_stub_factory,
Arc::new(ntk_qspn::FixedThreshold::default()),
Arc::new(ntk_qspn::DefaultArcIdSource::default()),
Vec::new(),
Vec::new(),
guest_gnode_level,
host_gnode_level,
(0, 0),
Vec::new(),
cancel.child_token(),
)?,
};
let qspn_events = qspn.subscribe_events();
tasks.spawn(async move {
let _ = qspn_join.await;
});
let svc = services::spawn(
topology,
my_naddr.clone(),
hooking,
qspn.clone(),
registry.clone(),
links.clone(),
net.clone(),
tasks,
&cancel,
signing_key,
require_auth,
coordinator_handoff,
)
.await;
let mut installer = RouteInstaller::new(kernel, route_naddr, table, rule_priority);
installer.install_identity().await?;
let arc_resolver = Arc::new(QspnArcResolverAdapter {
registry: registry.clone(),
});
let qspn_cfg = ntk_qspn::QspnConfig::default();
let qspn_rpc = ntk_qspn::QspnRpcHandler::new(
qspn.clone(),
arc_resolver,
qspn_cfg.arc_timeout,
qspn_cfg.caller_arc_poll_interval,
);
let peers_rpc = ntk_peerservices::PeersRpcHandler::new(svc.peers.clone());
let coordinator_rpc = ntk_coordinator::CoordinatorRpcHandler::new(svc.coordinator.clone());
let hooking_stub_factory = Arc::new(HookingStubFactoryAdapter {
qspn: qspn.clone(),
links: links.clone(),
registry,
});
let coord_client_for_router = Arc::new(CoordinatorClientAdapter::new(
svc.coordinator_client.clone(),
svc.coordinator.clone(),
qspn.clone(),
net,
services::coordinator_config().n_nodes_cache_ttl,
));
let view_for_router = svc.qspn_view.clone() as Arc<dyn ntk_hooking::QspnView>;
let router = Arc::new(ntk_hooking::MessageRouting::new(
view_for_router.clone(),
coord_client_for_router.clone(),
hooking_stub_factory,
ntk_hooking::HookingConfig::default().routing_response_timeout,
));
let hooking_rpc =
ntk_hooking::HookingRpcHandler::new(view_for_router, coord_client_for_router, router);
Ok(Generation {
qspn,
qspn_events,
hooking: svc.hooking,
peers: svc.peers,
coordinator: svc.coordinator,
andna: svc.andna,
dispatch: IdentityStack {
qspn: qspn_rpc,
peers: peers_rpc,
coordinator: coordinator_rpc,
hooking: hooking_rpc,
},
route_installer: installer,
})
}
pub async fn run<K>(
inputs: NodeInputs<K>,
tasks: &mut JoinSet<()>,
cancel: CancellationToken,
) -> anyhow::Result<StartedNode<K>>
where
K: SendNetlink + 'static,
{
let NodeInputs {
config,
neighborhood,
registry,
links,
routing_kernel,
dialer,
initial_position,
preformed,
my_id,
} = inputs;
anyhow::ensure!(
initial_position.is_none() || preformed.is_none(),
"NodeInputs::initial_position and NodeInputs::preformed are mutually exclusive: \
initial_position={initial_position:?} preformed={preformed:?}"
);
links.set_port(config.port());
let signing_key = config
.node_key_path()
.map(crate::node::andna_key::load_or_generate)
.transpose()?;
let require_auth = config.require_auth();
let kernel = KernelHandle(routing_kernel.clone());
crate::kernel::preflight::check(&kernel).await?;
let topology = config.topology()?;
let levels = topology.levels();
let negotiated = initial_position.is_none();
let origin = ntk_hooking::HookingOrigin::CreateNet;
let position = match (initial_position, &preformed) {
(Some(pos), _) => pos,
(None, Some(preformed)) => preformed.position.clone(),
(None, None) => derive_initial_position(my_id, &topology),
};
let my_naddr = Naddr::new(topology.clone(), position)?;
let network_id = preformed.map_or_else(random_i64, |preformed| preformed.network_id);
let net = Arc::new(NetworkInfo::new(levels, network_id));
let identity_stub_factory = Arc::new(IdentityStubFactoryAdapter {
links: links.clone(),
registry: registry.clone(),
});
let (identities, identities_join) = ntk_identities::Handle::spawn(
Some(my_naddr.clone()),
identity_stub_factory.clone(),
cancel.child_token(),
);
tasks.spawn(async move {
let _ = identities_join.await;
});
let table_allocator: TableAllocator<MigrationId> = TableAllocator::new();
let generation_cancel = cancel.child_token();
let generation = bootstrap_generation(
topology.clone(),
my_naddr.clone(),
QspnOrigin::CreateNet,
HookingProvenance::Fresh(origin),
net.clone(),
registry.clone(),
links.clone(),
my_id,
kernel,
table_allocator.main_table(),
table_allocator.main_rule_priority(),
tasks,
generation_cancel.clone(),
signing_key.clone(),
require_auth,
None,
)
.await?;
let table = table_allocator.main_table();
let generation_handles = GenerationHandles::from_generation(&generation, 0);
let identity_rpc =
ntk_identities::IdentityRpcHandler::new(identities.clone(), identity_stub_factory);
let neighborhood_rpc =
ntk_neighborhood::NeighborhoodRpcHandler::for_unicast(neighborhood.clone());
let dispatcher = Arc::new(Dispatcher::new(
neighborhood_rpc,
identity_rpc,
identities.clone(),
generation.dispatch,
));
let route_installer = Arc::new(tokio::sync::Mutex::new(generation.route_installer));
let net_for_running = net.clone();
let (generation_handles_tx, generation_handles_rx) = watch::channel(generation_handles);
tasks.spawn(run_steady_state(
SteadyStateCtx {
config_port: config.port(),
topology,
my_id,
qspn: generation.qspn.clone(),
identities: identities.clone(),
neighborhood: neighborhood.clone(),
hooking: generation.hooking.clone(),
peers: generation.peers.clone(),
registry: registry.clone(),
links,
dialer,
net,
route_installer: route_installer.clone(),
kernel: routing_kernel.clone(),
dispatcher: dispatcher.clone(),
generation_handles_tx,
negotiated,
migration_in_progress: false,
migrations: 0,
generation_cancel,
generation_tasks: JoinSet::new(),
signing_key,
require_auth,
table_allocator,
},
generation.qspn_events,
cancel,
));
Ok(StartedNode {
running: RunningNode {
generation: generation_handles_rx,
identities,
neighborhood,
hooking: generation.hooking,
registry,
route_table: table,
route_installer,
kernel: routing_kernel,
net: net_for_running,
},
dispatcher,
})
}
fn random_i64() -> i64 {
let raw = RandomState::new().build_hasher().finish();
(raw & 0x7fff_ffff_ffff_ffff).max(1) as i64
}
fn random_fingerprint_id() -> Vec<u8> {
RandomState::new()
.build_hasher()
.finish()
.to_be_bytes()
.to_vec()
}
fn random_migration_id() -> MigrationId {
let raw = RandomState::new().build_hasher().finish() as i32;
MigrationId(raw.checked_abs().unwrap_or(i32::MAX).max(1))
}
struct SteadyStateCtx<K> {
config_port: u16,
topology: Topology,
my_id: ntk_neighborhood::NodeId,
qspn: ntk_qspn::QspnHandle,
identities: ntk_identities::Handle,
neighborhood: ntk_neighborhood::Handle,
hooking: ntk_hooking::HookingHandle,
peers: ntk_peerservices::Handle,
registry: Arc<LinkRegistry>,
links: Arc<PeerLinks>,
dialer: Arc<dyn Dialer>,
net: Arc<NetworkInfo>,
route_installer: Arc<tokio::sync::Mutex<RouteInstaller<KernelHandle<K>>>>,
kernel: Arc<K>,
dispatcher: Arc<Dispatcher>,
generation_handles_tx: watch::Sender<GenerationHandles>,
negotiated: bool,
migration_in_progress: bool,
migrations: u32,
generation_cancel: CancellationToken,
generation_tasks: JoinSet<()>,
signing_key: Option<ed25519_dalek::SigningKey>,
require_auth: bool,
table_allocator: TableAllocator<MigrationId>,
}
async fn run_steady_state<K>(
mut ctx: SteadyStateCtx<K>,
mut qspn_events: broadcast::Receiver<QspnEvent>,
cancel: CancellationToken,
) where
K: SendNetlink + 'static,
{
let mut neighborhood_events = ctx.neighborhood.subscribe();
let mut hooking_events = ctx.hooking.subscribe_events();
loop {
tokio::select! {
biased;
() = cancel.cancelled() => break,
event = neighborhood_events.recv() => {
match event {
Ok(ev) => on_neighborhood_event(&ctx, ev).await,
Err(broadcast::error::RecvError::Closed) => break,
Err(broadcast::error::RecvError::Lagged(_)) => {}
}
}
event = qspn_events.recv() => {
match event {
Ok(ev) => on_qspn_event(&ctx, ev).await,
Err(broadcast::error::RecvError::Closed) => break,
Err(broadcast::error::RecvError::Lagged(_)) => on_route_snapshot_changed(&ctx).await,
}
}
event = hooking_events.recv() => {
match event {
Ok(ev) => {
if let Some(new_qspn_events) = on_hooking_event(&mut ctx, &cancel, ev).await
{
qspn_events = new_qspn_events;
}
}
Err(broadcast::error::RecvError::Closed) => break,
Err(broadcast::error::RecvError::Lagged(_)) => {}
}
}
}
}
while ctx.generation_tasks.join_next().await.is_some() {}
}
async fn reattach_known_arcs<K>(
ctx: &SteadyStateCtx<K>,
installer: &mut RouteInstaller<KernelHandle<K>>,
) where
K: SendNetlink + 'static,
{
let arcs = ctx.neighborhood.snapshot().borrow().clone();
let mut reattached_any = false;
for arc in arcs {
let Some(cost) = arc.cost else { continue };
let link =
ctx.registry
.link_for_neighbour(arc.neighbour_id, &arc.neighbour_mac, &arc.my_dev);
let qspn_arc = match ctx.qspn.add_arc(cost).await {
Ok(qspn_arc) => qspn_arc,
Err(err) => {
tracing::warn!(?link, %err, "qspn: re-attaching a known arc to the negotiated generation failed");
continue;
}
};
ctx.registry.set_qspn_arc(link, qspn_arc);
if let Ok(via) = arc.neighbour_nic_addr.parse() {
installer.set_arc_endpoint(qspn_arc, via, Interface::name(&arc.my_dev));
}
reattached_any = true;
}
if reattached_any {
let peers = ctx.peers.clone();
tokio::spawn(async move {
peers.reannounce_participation().await;
});
}
}
const BAD_LINK_RETRY_MIN_INTERVAL: std::time::Duration = std::time::Duration::from_millis(500);
async fn retry_removed_arc<K>(ctx: &SteadyStateCtx<K>, arc: ntk_qspn::ArcId)
where
K: SendNetlink + 'static,
{
let Some(link) = ctx.registry.link_of_qspn_arc(arc) else {
return;
};
if !ctx
.registry
.should_retry_bad_link(link, BAD_LINK_RETRY_MIN_INTERVAL)
{
return;
}
let Some(entry) = ctx.registry.entry(link) else {
return;
};
let arcs = ctx.neighborhood.snapshot().borrow().clone();
let Some(current) = arcs.iter().find(|a| a.neighbour_mac == entry.mac) else {
return;
};
let Some(cost) = current.cost else {
return;
};
let new_arc = match ctx.qspn.add_arc(cost).await {
Ok(new_arc) => new_arc,
Err(err) => {
tracing::warn!(?link, %err, "qspn: retrying a bad-link arc failed");
return;
}
};
tracing::debug!(?link, ?cost, "qspn: retried a bad-link arc, succeeded");
ctx.registry.set_qspn_arc(link, new_arc);
if let Ok(via) = current.neighbour_nic_addr.parse() {
ctx.route_installer.lock().await.set_arc_endpoint(
new_arc,
via,
Interface::name(¤t.my_dev),
);
}
}
#[must_use]
fn combine_entry_naddr(
topology: &Topology,
current_positions: &[u32],
guest_gnode_level: usize,
entry_pos: &[u32],
) -> Option<Naddr> {
let prefix = current_positions.get(..guest_gnode_level)?;
let mut pos = prefix.to_vec();
pos.extend_from_slice(entry_pos);
Naddr::new(topology.clone(), pos).ok()
}
#[must_use]
fn is_stale_finish_enter_replay(
current_naddr: &Naddr,
current_network_id: i64,
new_naddr: &Naddr,
target_network_id: i64,
) -> bool {
current_naddr == new_naddr && current_network_id == target_network_id
}
async fn migrate<K>(
ctx: &mut SteadyStateCtx<K>,
cancel: &CancellationToken,
guest_gnode_level: usize,
new_naddr: Naddr,
target_network_id: i64,
) -> Option<broadcast::Receiver<QspnEvent>>
where
K: SendNetlink + 'static,
{
if !ctx.negotiated || ctx.migration_in_progress {
return None;
}
if is_stale_finish_enter_replay(
ctx.qspn.my_naddr(),
ctx.net.network_id(),
&new_naddr,
target_network_id,
) {
tracing::debug!(
?new_naddr,
target_network_id,
"hooking: finish_enter re-proposed the position already held, ignoring stale replay"
);
return None;
}
let host_gnode_level = ctx.topology.levels();
if guest_gnode_level >= host_gnode_level {
tracing::warn!(
guest_gnode_level,
host_gnode_level,
"hooking: finish_enter's guest_gnode_level doesn't span every topology level, \
staying at the current position (not modeled by this daemon)"
);
return None;
}
ctx.migration_in_progress = true;
ctx.net.set_network_id(target_network_id);
tracing::info!(
?new_naddr,
guest_gnode_level,
target_network_id,
"hooking: migrating"
);
let old_id = ctx.identities.main_id();
let migration_id = random_migration_id();
if let Err(err) = ctx.identities.prepare_migration(migration_id, old_id).await {
tracing::warn!(%err, "identities: prepare_migration failed, staying at the current position");
ctx.migration_in_progress = false;
return None;
}
let new_id = match ctx
.identities
.migrate(migration_id, old_id, HashMap::new())
.await
{
Ok(id) => id,
Err(err) => {
tracing::warn!(%err, "identities: migrate failed, staying at the current position");
ctx.migration_in_progress = false;
return None;
}
};
if let Err(err) = ctx
.identities
.set_naddr(new_id, Some(virtual_placeholder(&ctx.topology)))
.await
{
tracing::warn!(%err, "identities: set_naddr(virtual) on the migrating identity failed");
}
let outgoing_coordinator = ctx.generation_handles_tx.borrow().coordinator.clone();
let coordinator_handoff = outgoing_coordinator.hand_off().await;
ctx.generation_cancel.cancel();
while ctx.generation_tasks.join_next().await.is_some() {}
if let Err(err) = ctx.route_installer.lock().await.teardown().await {
tracing::warn!(%err, "route installer: teardown of the previous generation's kernel state failed");
}
let new_generation_cancel = cancel.child_token();
let mut new_tasks = JoinSet::new();
let generation = match bootstrap_generation(
ctx.topology.clone(),
new_naddr.clone(),
QspnOrigin::Entering {
guest_gnode_level,
host_gnode_level,
},
HookingProvenance::Carried(ctx.hooking.clone()),
ctx.net.clone(),
ctx.registry.clone(),
ctx.links.clone(),
ctx.my_id,
KernelHandle(ctx.kernel.clone()),
ctx.table_allocator.main_table(),
ctx.table_allocator.main_rule_priority(),
&mut new_tasks,
new_generation_cancel.clone(),
ctx.signing_key.clone(),
ctx.require_auth,
Some(coordinator_handoff),
)
.await
{
Ok(generation) => generation,
Err(err) => {
tracing::error!(%err, "migrate: bootstrapping the entering generation failed, aborting");
if let Err(abort_err) = ctx.identities.abort_migration(old_id, new_id).await {
tracing::warn!(%abort_err, "identities: abort_migration after a failed entering-generation bootstrap failed");
}
ctx.migration_in_progress = false;
return None;
}
};
ctx.generation_cancel = new_generation_cancel;
ctx.generation_tasks = new_tasks;
let new_handles = GenerationHandles::from_generation(&generation, ctx.migrations + 1);
ctx.qspn = generation.qspn;
let new_qspn_events = generation.qspn_events;
ctx.hooking = generation.hooking;
ctx.peers = generation.peers;
let mut new_installer = generation.route_installer;
reattach_known_arcs(ctx, &mut new_installer).await;
*ctx.route_installer.lock().await = new_installer;
ctx.generation_handles_tx.send_replace(new_handles);
ctx.dispatcher
.replace_identity_stack(generation.dispatch)
.await;
loop {
match ctx.qspn.is_bootstrap_complete().await {
Ok(true) | Err(_) => break,
Ok(false) => tokio::time::sleep(MIGRATION_POLL_INTERVAL).await,
}
}
{
let mut installer = ctx.route_installer.lock().await;
installer.realize(new_naddr.clone());
if let Err(err) = installer.install_identity().await {
tracing::warn!(%err, "route installer: install_identity after migrate failed");
}
}
on_route_snapshot_changed(ctx).await;
if let Err(err) = ctx.identities.set_naddr(new_id, Some(new_naddr)).await {
tracing::warn!(%err, "identities: set_naddr(real) after migrate failed");
}
if let Err(err) = ctx.identities.remove_identity(old_id).await {
tracing::warn!(%err, "identities: remove_identity for the superseded connectivity fork failed");
}
ctx.migrations += 1;
ctx.migration_in_progress = false;
Some(new_qspn_events)
}
const MIGRATION_POLL_INTERVAL: Duration = Duration::from_millis(20);
async fn on_neighborhood_event<K>(ctx: &SteadyStateCtx<K>, event: ntk_neighborhood::Event)
where
K: SendNetlink + 'static,
{
match event {
ntk_neighborhood::Event::ArcAdded(arc) => {
let link =
ctx.registry
.link_for_neighbour(arc.neighbour_id, &arc.neighbour_mac, &arc.my_dev);
if ctx.links.get(link).is_none()
&& let Some(client) = ctx
.dialer
.dial_via(&arc.neighbour_nic_addr, ctx.config_port, Some(&arc.my_dev))
.await
{
ctx.links.insert(link, client);
}
let Some(cost) = arc.cost else { return };
let add_arc_result = ctx.qspn.add_arc(cost).await;
tracing::debug!(?link, ?cost, mac = %arc.neighbour_mac, ok = add_arc_result.is_ok(), "qspn: add_arc for newly discovered arc");
let qspn_arc = match add_arc_result {
Ok(qspn_arc) => qspn_arc,
Err(err) => {
tracing::warn!(?link, mac = %arc.neighbour_mac, %err, "qspn: add_arc failed, arc will not be routed");
return;
}
};
ctx.registry.set_qspn_arc(link, qspn_arc);
if let Ok(via) = arc.neighbour_nic_addr.parse() {
ctx.route_installer.lock().await.set_arc_endpoint(
qspn_arc,
via,
Interface::name(&arc.my_dev),
);
}
let _ = ctx
.identities
.add_arc(
link.identities(),
ArcInfo {
dev: arc.my_dev.clone(),
peer_mac: arc.neighbour_mac.clone(),
peer_linklocal: arc.neighbour_nic_addr.clone(),
},
)
.await;
let _ = ctx.hooking.add_arc(link.hooking()).await;
let peers = ctx.peers.clone();
tokio::spawn(async move {
peers.reannounce_participation().await;
});
on_route_snapshot_changed(ctx).await;
}
ntk_neighborhood::Event::ArcCostChanged(arc) => {
let Some(link) = ctx.registry.link_for_dev_and_mac(&arc.neighbour_mac) else {
return;
};
let Some(qspn_arc) = ctx.registry.qspn_arc_of(link) else {
return;
};
let Some(cost) = arc.cost else { return };
let _ = ctx.qspn.arc_changed(qspn_arc, cost).await;
on_route_snapshot_changed(ctx).await;
}
ntk_neighborhood::Event::ArcRemoved(arc) => {
if let Some(entry) = ctx.registry.remove(&arc.neighbour_mac) {
if let Some(qspn_arc) = entry.qspn_arc {
let _ = ctx.qspn.remove_arc(qspn_arc).await;
ctx.route_installer
.lock()
.await
.clear_arc_endpoint(qspn_arc);
}
let _ = ctx.identities.remove_arc(entry.id.identities()).await;
let _ = ctx.hooking.remove_arc(entry.id.hooking()).await;
ctx.links.remove(entry.id);
}
on_route_snapshot_changed(ctx).await;
}
}
}
async fn on_qspn_event<K>(ctx: &SteadyStateCtx<K>, event: QspnEvent)
where
K: SendNetlink + 'static,
{
match event {
QspnEvent::BootstrapComplete => ctx.net.set_bootstrapped(),
QspnEvent::DestinationAdded(_)
| QspnEvent::DestinationRemoved(_)
| QspnEvent::PathAdded(_)
| QspnEvent::PathChanged(_)
| QspnEvent::PathRemoved(_) => on_route_snapshot_changed(ctx).await,
QspnEvent::GnodeSplitted { destination, .. } => {
tracing::info!(
?destination,
"qspn: g-node split observed (not further modeled by this daemon)"
);
}
QspnEvent::PresenceNotified
| QspnEvent::ArcRemoved {
bad_link: false, ..
}
| QspnEvent::ChangedFingerprint(_)
| QspnEvent::ChangedNodesInside(_) => {}
QspnEvent::ArcRemoved {
arc,
bad_link: true,
} => retry_removed_arc(ctx, arc).await,
}
}
async fn on_route_snapshot_changed<K>(ctx: &SteadyStateCtx<K>)
where
K: SendNetlink + 'static,
{
let snapshot = ctx.qspn.snapshot();
let mut installer = ctx.route_installer.lock().await;
if let Err(err) = installer.apply(&snapshot).await {
tracing::warn!(%err, "route installer: apply failed");
}
}
async fn on_hooking_event<K>(
ctx: &mut SteadyStateCtx<K>,
cancel: &CancellationToken,
event: HookingEvent,
) -> Option<broadcast::Receiver<QspnEvent>>
where
K: SendNetlink + 'static,
{
match event {
HookingEvent::FailingArc(arc) => tracing::warn!(?arc, "hooking: arc failing"),
HookingEvent::SameNetwork(arc) => tracing::debug!(?arc, "hooking: peer on same network"),
HookingEvent::AnotherNetwork { arc, network_id } => {
tracing::info!(
?arc,
network_id,
"hooking: peer belongs to another network, merge evaluation started"
);
}
HookingEvent::DoPrepareEnter { enter_id } => {
tracing::debug!(enter_id, "hooking: prepare_enter propagated");
}
HookingEvent::DoFinishEnter {
guest_gnode_level,
data,
} => {
tracing::debug!(
guest_gnode_level,
network_id = data.entry_data.network_id,
"hooking: finish_enter resolved, requesting migration"
);
let current = ctx.qspn.my_naddr().positions();
let new_naddr = combine_entry_naddr(
&ctx.topology,
current,
guest_gnode_level,
&data.entry_data.pos,
);
match new_naddr {
Some(new_naddr) => {
return migrate(
ctx,
cancel,
guest_gnode_level,
new_naddr,
data.entry_data.network_id,
)
.await;
}
None => tracing::warn!(
guest_gnode_level,
entry_pos = ?data.entry_data.pos,
current_pos = ?current,
"hooking: finish_enter's position doesn't combine into a valid address, staying at the current position"
),
}
}
HookingEvent::DoPrepareMigration { migration_id } => {
let old_id = ctx.identities.main_id();
let _ = ctx
.identities
.prepare_migration(ntk_identities::MigrationId(migration_id), old_id)
.await;
}
HookingEvent::DoFinishMigration { .. } => {
tracing::info!(
"hooking: finish_migration notified (identity re-hooking not modeled by this daemon)"
);
}
}
None
}
#[must_use]
pub fn synthetic_mac(dev: &str, my_id: ntk_neighborhood::NodeId) -> String {
use std::hash::{Hash, Hasher};
let mut hasher = std::collections::hash_map::DefaultHasher::new();
my_id.hash(&mut hasher);
dev.hash(&mut hasher);
let b = hasher.finish().to_be_bytes();
let first = (b[0] & 0xfc) | 0x02;
format!(
"{first:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}",
b[1], b[2], b[3], b[4], b[5]
)
}
#[cfg(test)]
mod synthetic_mac_tests {
use super::synthetic_mac;
use ntk_neighborhood::NodeId;
fn assert_locally_administered_unicast(mac: &str) {
let first = u8::from_str_radix(mac.split(':').next().unwrap(), 16).unwrap();
assert_eq!(
first & 0x02,
0x02,
"{mac}: locally-administered bit not set"
);
assert_eq!(first & 0x01, 0, "{mac}: multicast bit not clear");
assert_eq!(mac.split(':').count(), 6, "{mac}: not 6 octets");
}
#[test]
fn different_node_ids_diverge_on_the_same_device_name() {
let a = synthetic_mac("eth0", NodeId::from_raw(1).unwrap());
let b = synthetic_mac("eth0", NodeId::from_raw(2).unwrap());
assert_ne!(a, b);
}
#[test]
fn same_node_and_device_is_deterministic() {
let id = NodeId::from_raw(42).unwrap();
assert_eq!(synthetic_mac("eth0", id), synthetic_mac("eth0", id));
}
#[test]
fn different_devices_on_the_same_node_diverge() {
let id = NodeId::from_raw(7).unwrap();
assert_ne!(synthetic_mac("eth0", id), synthetic_mac("eth1", id));
}
#[test]
fn result_is_always_a_valid_locally_administered_unicast_mac() {
for raw in [1, 2, 3, 42, i32::MAX] {
for dev in ["eth0", "wlan0", "enp0s3"] {
assert_locally_administered_unicast(&synthetic_mac(
dev,
NodeId::from_raw(raw).unwrap(),
));
}
}
}
}
#[cfg(test)]
mod linklocal_allocator_tests {
use super::{LINKLOCAL_SPACE, derive_linklocal};
use ntk_neighborhood::NodeId;
use std::net::Ipv4Addr;
fn assert_valid_linklocal(addr: Ipv4Addr) {
let o = addr.octets();
assert_eq!(&o[0..2], [169, 254], "{addr}: not inside 169.254.0.0/16");
assert_ne!(o[2], 0, "{addr}: inside reserved 169.254.0.0/24");
assert_ne!(o[2], 255, "{addr}: inside reserved 169.254.255.0/24");
}
#[test]
fn different_node_ids_diverge_on_the_same_nic_index() {
let a = derive_linklocal(NodeId::from_raw(1).unwrap(), 0);
let b = derive_linklocal(NodeId::from_raw(2).unwrap(), 0);
assert_ne!(a, b);
}
#[test]
fn same_salt_and_index_is_deterministic() {
let id = NodeId::from_raw(42).unwrap();
assert_eq!(derive_linklocal(id, 3), derive_linklocal(id, 3));
}
#[test]
fn different_indices_on_the_same_node_diverge() {
let id = NodeId::from_raw(7).unwrap();
assert_ne!(derive_linklocal(id, 0), derive_linklocal(id, 1));
assert_ne!(derive_linklocal(id, 0), derive_linklocal(id, 41));
}
#[test]
fn every_generated_address_stays_inside_the_valid_linklocal_range() {
for raw in [1, 2, 3, 42, i32::MAX] {
let id = NodeId::from_raw(raw).unwrap();
for index in [0, 1, 2, 100, LINKLOCAL_SPACE - 1] {
assert_valid_linklocal(derive_linklocal(id, index));
}
}
}
}
#[cfg(test)]
mod combine_entry_naddr_tests {
use super::combine_entry_naddr;
use ntk_common::Topology;
fn topo() -> Topology {
Topology::new([4, 2]).expect("valid topology")
}
#[test]
fn level_zero_is_the_entry_pos_verbatim_regardless_of_current_position() {
for current in [[0, 0], [3, 1], [2, 0]] {
let naddr = combine_entry_naddr(&topo(), ¤t, 0, &[1, 0])
.expect("full-length entry_pos at level 0 is always valid");
assert_eq!(naddr.positions(), [1, 0]);
}
}
#[test]
fn three_members_with_distinct_lower_positions_share_one_propagated_upper_position() {
let entry_pos = [1]; let members = [
combine_entry_naddr(&topo(), &[0, 0], 1, &entry_pos).expect("member 0 combines"),
combine_entry_naddr(&topo(), &[1, 0], 1, &entry_pos).expect("member 1 combines"),
combine_entry_naddr(&topo(), &[2, 0], 1, &entry_pos).expect("member 2 combines"),
];
assert!(members.iter().all(|n| n.positions()[1] == 1));
let level0: std::collections::HashSet<u32> =
members.iter().map(|n| n.positions()[0]).collect();
assert_eq!(level0, std::collections::HashSet::from([0, 1, 2]));
let distinct: std::collections::HashSet<&[u32]> =
members.iter().map(ntk_common::Naddr::positions).collect();
assert_eq!(distinct.len(), 3);
}
#[test]
fn out_of_range_guest_gnode_level_is_rejected_not_panicked() {
assert!(combine_entry_naddr(&topo(), &[0, 0], 3, &[]).is_none());
}
#[test]
fn wrong_total_length_is_rejected() {
assert!(combine_entry_naddr(&topo(), &[0, 0], 1, &[1, 0]).is_none());
assert!(combine_entry_naddr(&topo(), &[0, 0], 1, &[]).is_none());
}
#[test]
fn out_of_bounds_position_is_rejected() {
assert!(combine_entry_naddr(&topo(), &[0, 0], 1, &[5]).is_none());
}
}
#[cfg(test)]
mod is_stale_finish_enter_replay_tests {
use super::is_stale_finish_enter_replay;
use ntk_common::{Naddr, Topology};
fn topo() -> Topology {
Topology::new([4, 2]).expect("valid topology")
}
fn naddr(pos: [u32; 2]) -> Naddr {
Naddr::new(topo(), pos.to_vec()).expect("valid address")
}
#[test]
fn same_position_and_network_is_stale() {
assert!(is_stale_finish_enter_replay(
&naddr([1, 0]),
42,
&naddr([1, 0]),
42
));
}
#[test]
fn same_position_different_network_is_not_stale() {
assert!(!is_stale_finish_enter_replay(
&naddr([1, 0]),
42,
&naddr([1, 0]),
99
));
}
#[test]
fn different_position_same_network_is_not_stale() {
assert!(!is_stale_finish_enter_replay(
&naddr([1, 0]),
42,
&naddr([2, 0]),
42
));
}
#[test]
fn different_position_and_network_is_not_stale() {
assert!(!is_stale_finish_enter_replay(
&naddr([1, 0]),
42,
&naddr([2, 0]),
99
));
}
}