use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use futures::future::BoxFuture;
use ntk_common::{HCoord, Naddr, Topology};
use ntk_hooking::{
CoordinatorClient as HookingCoordinatorClient, CoordinatorError, EvaluateEnterRequest,
FinishEnterData, FinishMigrationData, HookingHandle, MergeArbitrationRequest,
Reservation as HookingReservation, merge_tiebreak,
};
use ntk_peerservices::{PeersStub, RoutingEnv, TupleNode};
use ntk_proto::v1::TypedValue;
use ntk_qspn::{ArcId as QspnArcId, QspnEvent, QspnHandle, RouteSnapshot};
use crate::node::codec;
use crate::node::peers::PeerLinks;
use crate::node::registry::{LinkId, LinkRegistry};
#[derive(Debug)]
pub struct NetworkInfo {
network_id: AtomicI64,
bootstrapped: std::sync::atomic::AtomicBool,
foreign: Mutex<HashSet<(usize, u32)>>,
same_network: Mutex<HashSet<(usize, u32)>>,
}
impl NetworkInfo {
#[must_use]
pub fn new(_levels: usize, initial_network_id: i64) -> Self {
Self {
network_id: AtomicI64::new(initial_network_id),
bootstrapped: std::sync::atomic::AtomicBool::new(false),
foreign: Mutex::new(HashSet::new()),
same_network: Mutex::new(HashSet::new()),
}
}
pub fn set_network_id(&self, id: i64) {
self.network_id.store(id, Ordering::Relaxed);
}
#[must_use]
pub fn network_id(&self) -> i64 {
self.network_id.load(Ordering::Relaxed)
}
pub fn note_foreign(&self, level: usize, pos: u32) {
self.foreign
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert((level, pos));
}
pub fn note_same_network(&self, level: usize, pos: u32) {
self.same_network
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert((level, pos));
}
#[must_use]
pub fn is_foreign(&self, level: usize, pos: u32) -> bool {
if self
.same_network
.lock()
.unwrap_or_else(|e| e.into_inner())
.contains(&(level, pos))
{
return false;
}
self.foreign
.lock()
.unwrap_or_else(|e| e.into_inner())
.contains(&(level, pos))
}
#[must_use]
pub fn foreign_positions(&self) -> Vec<(usize, u32)> {
let same_network = self.same_network.lock().unwrap_or_else(|e| e.into_inner());
self.foreign
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter()
.filter(|entry| !same_network.contains(entry))
.copied()
.collect()
}
pub fn set_bootstrapped(&self) {
self.bootstrapped.store(true, Ordering::Relaxed);
}
#[must_use]
pub fn is_bootstrapped(&self) -> bool {
self.bootstrapped.load(Ordering::Relaxed)
}
pub fn reset_for_new_generation(&self) {
self.bootstrapped.store(false, Ordering::Relaxed);
self.foreign
.lock()
.unwrap_or_else(|e| e.into_inner())
.clear();
self.same_network
.lock()
.unwrap_or_else(|e| e.into_inner())
.clear();
}
}
type EldershipSlot = Option<Option<u32>>;
#[derive(Debug, Default)]
struct EldershipCache {
my: Mutex<HashMap<usize, EldershipSlot>>,
foreign: Mutex<HashMap<(usize, u32), EldershipSlot>>,
}
fn map_eldership(v: Option<Option<u32>>) -> i32 {
match v {
Some(Some(n)) => i32::try_from(n).unwrap_or(i32::MAX - 1),
Some(None) => -1,
None => i32::MAX,
}
}
impl EldershipCache {
fn my(&self, level: usize) -> i32 {
map_eldership(
self.my
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(&level)
.copied()
.flatten(),
)
}
fn foreign(&self, level: usize, pos: u32) -> i32 {
map_eldership(
self.foreign
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(&(level, pos))
.copied()
.flatten(),
)
}
fn set_my(&self, level: usize, value: Option<Option<u32>>) {
self.my
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(level, value);
}
fn set_foreign(&self, level: usize, pos: u32, value: Option<Option<u32>>) {
self.foreign
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert((level, pos), value);
}
fn forget_foreign(&self, level: usize, pos: u32) {
self.foreign
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&(level, pos));
}
fn retain_foreign(&self, keep: impl Fn(&(usize, u32)) -> bool) {
self.foreign
.lock()
.unwrap_or_else(|e| e.into_inner())
.retain(|key, _| keep(key));
}
}
#[derive(Debug, Default, PartialEq, Eq)]
struct RefreshPlan {
full: bool,
destinations: HashSet<HCoord>,
}
impl RefreshPlan {
fn everything() -> Self {
Self {
full: true,
destinations: HashSet::new(),
}
}
fn is_noop(&self) -> bool {
!self.full && self.destinations.is_empty()
}
fn absorb(&mut self, event: &QspnEvent) {
match event {
QspnEvent::BootstrapComplete
| QspnEvent::PresenceNotified
| QspnEvent::ChangedFingerprint(_)
| QspnEvent::ArcRemoved { .. } => self.full = true,
QspnEvent::ChangedNodesInside(_) => {}
QspnEvent::DestinationAdded(destination)
| QspnEvent::DestinationRemoved(destination)
| QspnEvent::GnodeSplitted { destination, .. } => {
self.destinations.insert(*destination);
}
QspnEvent::PathAdded(path)
| QspnEvent::PathChanged(path)
| QspnEvent::PathRemoved(path) => match path.hops.last() {
Some(last) => {
self.destinations.insert(last.coord);
}
None => self.full = true,
},
}
}
fn absorb_pending(&mut self, events: &mut tokio::sync::broadcast::Receiver<QspnEvent>) {
use tokio::sync::broadcast::error::TryRecvError;
loop {
match events.try_recv() {
Ok(event) => self.absorb(&event),
Err(TryRecvError::Lagged(_)) => self.full = true,
Err(TryRecvError::Empty | TryRecvError::Closed) => return,
}
}
}
}
async fn refresh_eldership(
qspn: &QspnHandle,
levels: usize,
cache: &EldershipCache,
plan: &RefreshPlan,
) {
for level in 0..levels {
cache.set_my(level, qspn.my_eldership(level).await.unwrap_or(None));
}
let snapshot = qspn.snapshot();
let exported = |coord: HCoord| {
snapshot
.levels
.get(coord.level)
.is_some_and(|entries| entries.iter().any(|e| e.destination == coord))
};
if plan.full {
let mut keep = HashSet::new();
for (level, entries) in snapshot.levels.iter().enumerate() {
for entry in entries {
let pos = entry.destination.pos;
keep.insert((level, pos));
cache.set_foreign(level, pos, qspn.eldership(level, pos).await.unwrap_or(None));
}
}
cache.retain_foreign(|key| keep.contains(key));
return;
}
for &coord in &plan.destinations {
if exported(coord) {
cache.set_foreign(
coord.level,
coord.pos,
qspn.eldership(coord.level, coord.pos).await.unwrap_or(None),
);
} else {
cache.forget_foreign(coord.level, coord.pos);
}
}
}
async fn run_eldership_cache(
mut generation: tokio::sync::watch::Receiver<QspnHandle>,
cache: Arc<EldershipCache>,
cancel: tokio_util::sync::CancellationToken,
) {
loop {
let qspn = generation.borrow_and_update().clone();
let levels = qspn.my_naddr().topology().levels();
let mut events = qspn.subscribe_events();
let mut plan = RefreshPlan::everything();
loop {
if !plan.is_noop() {
refresh_eldership(&qspn, levels, &cache, &plan).await;
}
plan = RefreshPlan::default();
let closed = tokio::select! {
() = cancel.cancelled() => return,
changed = generation.changed() => {
if changed.is_err() {
return;
}
break;
}
event = events.recv() => match event {
Ok(event) => {
plan.absorb(&event);
plan.absorb_pending(&mut events);
false
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
plan.full = true;
false
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => true,
},
};
if closed {
tokio::select! {
() = cancel.cancelled() => return,
changed = generation.changed() => {
if changed.is_err() {
return;
}
break;
}
}
}
}
}
}
#[derive(Debug)]
pub struct Live<T>(Arc<tokio::sync::watch::Sender<T>>);
impl<T> Clone for Live<T> {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}
impl<T: Clone> Live<T> {
#[must_use]
pub fn new(value: T) -> Self {
Self(Arc::new(tokio::sync::watch::channel(value).0))
}
#[must_use]
pub fn get(&self) -> T {
self.0.borrow().clone()
}
pub fn set(&self, value: T) {
self.0.send_replace(value);
}
#[must_use]
pub fn subscribe(&self) -> tokio::sync::watch::Receiver<T> {
self.0.subscribe()
}
}
fn estimate_n_nodes(snapshot: &RouteSnapshot, net: Option<&NetworkInfo>, level: usize) -> u64 {
let mut counted: u64 = 0;
for entry in snapshot.levels.get(level).into_iter().flatten() {
if net.is_some_and(|n| n.is_foreign(entry.destination.level, entry.destination.pos)) {
continue;
}
let Some(path) = entry.paths.first() else {
continue;
};
counted += if level == 0 {
1
} else {
u64::from(path.nodes_inside)
};
}
let mine = if level == 0 {
1
} else {
estimate_n_nodes(snapshot, net, level - 1)
};
counted + mine
}
fn first_hop_link(
snapshot: &RouteSnapshot,
registry: &LinkRegistry,
hc: HCoord,
skip: Option<QspnArcId>,
) -> Option<LinkId> {
let entry = snapshot
.levels
.get(hc.level)?
.iter()
.find(|e| e.destination == hc)?;
entry
.paths
.iter()
.find(|p| Some(p.arc) != skip)
.and_then(|p| registry.link_of_qspn_arc(p.arc))
}
#[derive(Debug)]
pub struct QspnViewAdapter {
pub qspn: Live<QspnHandle>,
topology: Topology,
pub net: Arc<NetworkInfo>,
pub epsilon: usize,
eldership: Arc<EldershipCache>,
}
impl QspnViewAdapter {
#[must_use]
pub fn spawn(
qspn: QspnHandle,
net: Arc<NetworkInfo>,
tasks: &mut tokio::task::JoinSet<()>,
cancel: tokio_util::sync::CancellationToken,
) -> Self {
let topology = qspn.my_naddr().topology().clone();
let epsilon = hooking_epsilon(&topology);
let eldership = Arc::new(EldershipCache::default());
let qspn = Live::new(qspn);
tasks.spawn(run_eldership_cache(
qspn.subscribe(),
eldership.clone(),
cancel,
));
Self {
qspn,
topology,
net,
epsilon,
eldership,
}
}
pub fn retarget(&self, qspn: QspnHandle) {
self.qspn.set(qspn);
}
}
fn hooking_epsilon(topology: &Topology) -> usize {
let mut bits = 0u32;
for (count, gsize) in topology.gsizes().iter().enumerate() {
bits += 32 - gsize.saturating_sub(1).leading_zeros();
if bits >= 5 {
return count + 1;
}
}
topology.levels()
}
impl ntk_hooking::QspnView for QspnViewAdapter {
fn topology(&self) -> &Topology {
&self.topology
}
fn network_id(&self) -> i64 {
self.net.network_id()
}
fn n_nodes(&self) -> u64 {
estimate_n_nodes(
&self.qspn.get().snapshot(),
Some(&self.net),
self.topology().levels().saturating_sub(1),
)
}
fn my_pos(&self, level: usize) -> u32 {
self.qspn.get().my_naddr().pos(level).unwrap_or(0)
}
fn my_eldership(&self, level: usize) -> i32 {
self.eldership.my(level)
}
fn subnetlevel(&self) -> usize {
0
}
fn epsilon(&self, _level: usize) -> usize {
self.epsilon
}
fn eldership(&self, level: usize, pos: u32) -> i32 {
self.eldership.foreign(level, pos)
}
fn adjacent_to_my_gnode(
&self,
level_adjacent_gnodes: usize,
level_my_gnode: usize,
) -> Vec<ntk_hooking::AdjacentGNode> {
let qspn = self.qspn.get();
let snapshot = qspn.snapshot();
let my_naddr = qspn.my_naddr();
let Some(entries) = snapshot.levels.get(level_adjacent_gnodes) else {
return Vec::new();
};
entries
.iter()
.filter_map(|entry| {
let hc = entry.destination;
if my_naddr.is_inside(hc).unwrap_or(false) {
return None;
}
let border_real_pos = my_naddr.pos(level_my_gnode)?;
Some(ntk_hooking::AdjacentGNode {
hc,
border_real_pos,
})
})
.collect()
}
fn is_bootstrapped(&self) -> bool {
self.net.is_bootstrapped()
}
fn note_foreign(&self, level: usize, pos: u32) {
self.net.note_foreign(level, pos);
}
fn note_same_network(&self, level: usize, pos: u32) {
self.net.note_same_network(level, pos);
}
}
#[derive(Debug)]
pub struct CoordinatorClientAdapter {
pub dht: Live<ntk_coordinator::CoordinatorClient>,
pub local: Live<ntk_coordinator::Handle>,
pub qspn: Live<QspnHandle>,
pub net: Arc<NetworkInfo>,
merge_decisions: Mutex<HashMap<i64, (bool, Instant)>>,
merge_decision_ttl: Duration,
}
impl CoordinatorClientAdapter {
#[must_use]
pub fn new(
dht: ntk_coordinator::CoordinatorClient,
local: ntk_coordinator::Handle,
qspn: QspnHandle,
net: Arc<NetworkInfo>,
merge_decision_ttl: Duration,
) -> Self {
Self {
dht: Live::new(dht),
local: Live::new(local),
qspn: Live::new(qspn),
net,
merge_decisions: Mutex::new(HashMap::new()),
merge_decision_ttl,
}
}
pub fn retarget(
&self,
dht: ntk_coordinator::CoordinatorClient,
local: ntk_coordinator::Handle,
qspn: QspnHandle,
) {
self.dht.set(dht);
self.local.set(local);
self.qspn.set(qspn);
}
fn foreign_exclusions(&self) -> Vec<ntk_peerservices::TupleGNode> {
foreign_exclusions(&self.qspn.get(), &self.net)
}
}
fn foreign_exclusions(qspn: &QspnHandle, net: &NetworkInfo) -> Vec<ntk_peerservices::TupleGNode> {
let topology = qspn.my_naddr().topology().clone();
net.foreign_positions()
.into_iter()
.filter_map(|(level, pos)| {
ntk_peerservices::TupleGNode::new(topology.clone(), level + 1, vec![pos]).ok()
})
.collect()
}
fn proxy_err(e: ntk_coordinator::ProxyError) -> CoordinatorError {
CoordinatorError::Unreachable(e.to_string())
}
impl HookingCoordinatorClient for CoordinatorClientAdapter {
fn n_nodes(&self) -> BoxFuture<'_, u64> {
Box::pin(async move {
self.dht
.get()
.get_n_nodes(&self.foreign_exclusions())
.await
.unwrap_or(1)
})
}
fn evaluate_enter(
&self,
req: EvaluateEnterRequest,
) -> BoxFuture<'_, Result<usize, CoordinatorError>> {
Box::pin(async move {
let top = self.local.get().topology().levels();
let data = codec::encode_evaluate_enter_request(&req);
let reply = self
.dht
.get()
.evaluate_enter(top, data)
.await
.map_err(proxy_err)?;
let decoded = codec::decode_evaluate_enter_answer(&reply);
tracing::info!(
network_id = req.network_id,
evaluate_enter_id = req.evaluate_enter_id,
min_lvl = req.min_lvl,
?decoded,
"hooking client: evaluate_enter round trip"
);
match decoded {
Ok(codec::EvaluateEnterAnswer::Accepted { chosen_lvl }) => Ok(chosen_lvl),
Ok(codec::EvaluateEnterAnswer::AskAgain) => Err(CoordinatorError::AskAgain),
Ok(codec::EvaluateEnterAnswer::IgnoreNetwork) => {
Err(CoordinatorError::IgnoreNetwork)
}
Err(e) => Err(CoordinatorError::Unreachable(e.to_string())),
}
})
}
fn begin_enter(&self, lvl: usize) -> BoxFuture<'_, Result<(), CoordinatorError>> {
Box::pin(async move {
if lvl == 0 {
self.local
.get()
.begin_enter(1, codec::encode_unit(), &[])
.await;
return Ok(());
}
self.dht
.get()
.begin_enter(lvl + 1, codec::encode_unit())
.await
.map(drop)
.map_err(proxy_err)
})
}
fn completed_enter(&self, lvl: usize) -> BoxFuture<'_, Result<(), CoordinatorError>> {
Box::pin(async move {
if lvl == 0 {
self.local
.get()
.completed_enter(1, codec::encode_unit(), &[])
.await;
return Ok(());
}
self.dht
.get()
.completed_enter(lvl + 1, codec::encode_unit())
.await
.map(drop)
.map_err(proxy_err)
})
}
fn abort_enter(&self, lvl: usize) -> BoxFuture<'_, Result<(), CoordinatorError>> {
Box::pin(async move {
if lvl == 0 {
self.local
.get()
.abort_enter(1, codec::encode_unit(), &[])
.await;
return Ok(());
}
self.dht
.get()
.abort_enter(lvl + 1, codec::encode_unit())
.await
.map(drop)
.map_err(proxy_err)
})
}
fn reserve(
&self,
host_lvl: usize,
reserve_request_id: i32,
) -> BoxFuture<'_, Result<HookingReservation, CoordinatorError>> {
Box::pin(async move {
let exclude = self.foreign_exclusions();
let outcome = self
.dht
.get()
.reserve(host_lvl, i64::from(reserve_request_id), &exclude)
.await;
tracing::info!(
host_lvl,
reserve_request_id,
?exclude,
?outcome,
"coordinator: reserve outcome"
);
match outcome {
Ok(Some(r)) => Ok(HookingReservation {
pos: r.new_pos,
eldership: i32::try_from(r.new_eldership).unwrap_or(i32::MAX),
}),
Ok(None) => Err(CoordinatorError::NoCoordinatorForLevel),
Err(e) => Err(proxy_err(e)),
}
})
}
fn delete_reserve(&self, host_lvl: usize, reserve_request_id: i32) -> BoxFuture<'_, ()> {
Box::pin(async move {
let _ = self
.dht
.get()
.delete_reserve(
host_lvl,
i64::from(reserve_request_id),
&self.foreign_exclusions(),
)
.await;
})
}
fn prepare_migration(&self, lvl: usize, migration_id: i32) -> BoxFuture<'_, ()> {
Box::pin(async move {
self.local
.get()
.prepare_migration(lvl, codec::encode_migration_id(migration_id))
.await;
})
}
fn finish_migration(&self, lvl: usize, data: FinishMigrationData) -> BoxFuture<'_, ()> {
Box::pin(async move {
self.local
.get()
.finish_migration(lvl, codec::encode_finish_migration_data(&data))
.await;
})
}
fn prepare_enter(&self, lvl: usize, enter_id: i32) -> BoxFuture<'_, ()> {
Box::pin(async move {
tracing::info!(
lvl,
enter_id,
"migration-instrumentation: prepare_enter propagating"
);
self.local
.get()
.prepare_enter(lvl, codec::encode_enter_id(enter_id))
.await;
})
}
fn decide_merge(&self, req: MergeArbitrationRequest) -> BoxFuture<'_, bool> {
Box::pin(async move {
let now = Instant::now();
if let Some(&(cached, decided_at)) = self
.merge_decisions
.lock()
.expect("merge_decisions mutex poisoned")
.get(&req.neighbor_network_id)
&& now.saturating_duration_since(decided_at) < self.merge_decision_ttl
{
tracing::info!(
my_network_id = req.my_network_id,
neighbor_network_id = req.neighbor_network_id,
neighbor_n_nodes = req.neighbor_n_nodes,
decision = cached,
cached = true,
"migration-instrumentation: decide_merge"
);
return cached;
}
let top = self.local.get().topology().levels();
let now_ms = codec::now_millis();
let ttl_ms = u64::try_from(self.merge_decision_ttl.as_millis()).unwrap_or(u64::MAX);
let exclude = self.foreign_exclusions();
let mut mem = match self.dht.get().hooking_memory(top, &exclude).await {
Ok(Some(tv)) => codec::decode_hooking_memory(&tv).unwrap_or_default(),
_ => codec::HookingMemory::default(),
};
mem.merge_decisions
.retain(|_, &mut (_, decided_at_ms)| now_ms.saturating_sub(decided_at_ms) < ttl_ms);
let (decision, my_n_nodes) =
if let Some(&(decision, _)) = mem.merge_decisions.get(&req.neighbor_network_id) {
(decision, None)
} else {
let my_n_nodes = self.n_nodes().await;
let decision = merge_tiebreak(
my_n_nodes,
req.neighbor_n_nodes,
req.my_network_id,
req.neighbor_network_id,
);
mem.merge_decisions
.insert(req.neighbor_network_id, (decision, now_ms));
let _ = self
.dht
.get()
.set_hooking_memory(top, Some(codec::encode_hooking_memory(&mem)), &exclude)
.await;
(decision, Some(my_n_nodes))
};
tracing::info!(
my_network_id = req.my_network_id,
neighbor_network_id = req.neighbor_network_id,
neighbor_n_nodes = req.neighbor_n_nodes,
?my_n_nodes,
decision,
cached = false,
"migration-instrumentation: decide_merge"
);
self.merge_decisions
.lock()
.expect("merge_decisions mutex poisoned")
.insert(req.neighbor_network_id, (decision, now));
decision
})
}
fn finish_enter(&self, lvl: usize, data: FinishEnterData) -> BoxFuture<'_, ()> {
Box::pin(async move {
tracing::info!(
lvl,
enter_id = data.enter_id,
entry_network_id = data.entry_data.network_id,
entry_pos = ?data.entry_data.pos,
"migration-instrumentation: finish_enter propagating"
);
self.local
.get()
.finish_enter(lvl, codec::encode_finish_enter_data(&data))
.await;
})
}
}
#[derive(Debug)]
pub struct CoordinatorMapAdapter {
pub qspn: QspnHandle,
pub net: Arc<NetworkInfo>,
}
impl ntk_coordinator::CoordinatorMap for CoordinatorMapAdapter {
fn n_nodes(&self) -> u64 {
estimate_n_nodes(
&self.qspn.snapshot(),
Some(&self.net),
self.qspn.my_naddr().topology().levels().saturating_sub(1),
)
}
fn free_positions(&self, level: usize) -> Vec<u32> {
let topology = self.qspn.my_naddr().topology();
let Some(gsize) = topology.gsize(level) else {
return Vec::new();
};
let snapshot = self.qspn.snapshot();
let my_pos = self.qspn.my_naddr().pos(level).unwrap_or(0);
let raw_destinations: Vec<(usize, u32, bool)> = snapshot
.levels
.get(level)
.into_iter()
.flatten()
.map(|e| {
(
e.destination.level,
e.destination.pos,
self.net.is_foreign(e.destination.level, e.destination.pos),
)
})
.collect();
let mut occupied: std::collections::HashSet<u32> = snapshot
.levels
.get(level)
.into_iter()
.flatten()
.filter(|e| !self.net.is_foreign(e.destination.level, e.destination.pos))
.map(|e| e.destination.pos)
.collect();
occupied.insert(my_pos);
let free: Vec<u32> = (0..gsize).filter(|p| !occupied.contains(p)).collect();
tracing::info!(
level,
gsize,
my_pos,
network_id = self.net.network_id(),
?raw_destinations,
?occupied,
?free,
"coordinator: free_positions"
);
free
}
fn can_reserve(&self, level: usize) -> bool {
!self.free_positions(level).is_empty()
}
fn my_pos(&self, level: usize) -> u32 {
self.qspn.my_naddr().pos(level).unwrap_or(0)
}
fn fp_id(&self, _level: usize) -> i64 {
self.net.network_id()
}
}
#[derive(Debug, Default)]
struct LevelState {
elected: Option<Election>,
}
#[derive(Debug, Clone, Copy)]
struct Election {
network_id: i64,
evaluate_enter_id: i32,
granted_at: Instant,
completed_at: Option<Instant>,
}
const ELECTED_TTL: Duration = Duration::from_millis(60_000);
#[derive(Debug, Default)]
pub struct EnterArbiter {
levels: tokio::sync::Mutex<HashMap<usize, LevelState>>,
}
impl EnterArbiter {
#[must_use]
pub fn new() -> Self {
Self::default()
}
async fn decide(
&self,
service: &ntk_coordinator::CoordinatorService,
top: usize,
level: usize,
network_id: i64,
evaluate_enter_id: i32,
) -> codec::EvaluateEnterAnswer {
let mut levels = self.levels.lock().await;
let st = levels.entry(level).or_default();
if let Some(e) = st.elected {
let ttl_expired = e.granted_at.elapsed() >= ELECTED_TTL;
let released_for_other_networks =
e.completed_at.is_some() && e.network_id != network_id;
if ttl_expired || released_for_other_networks {
st.elected = None;
}
}
match st.elected {
Some(e) if e.network_id != network_id => {
return codec::EvaluateEnterAnswer::AskAgain;
}
Some(e) if e.evaluate_enter_id == evaluate_enter_id => {
return codec::EvaluateEnterAnswer::Accepted { chosen_lvl: level };
}
Some(_) => return codec::EvaluateEnterAnswer::IgnoreNetwork,
None => {}
}
let now_ms = codec::now_millis();
let ttl_ms = u64::try_from(ELECTED_TTL.as_millis()).unwrap_or(u64::MAX);
let mut mem = match service.hooking_memory_locally(top).await {
Some(tv) => codec::decode_hooking_memory(&tv).unwrap_or_default(),
None => codec::HookingMemory::default(),
};
let shared = mem.elections.get(&level).copied().filter(|e| {
let ttl_expired = now_ms.saturating_sub(e.granted_at_millis) >= ttl_ms;
let released_for_other_networks =
e.completed_at_millis.is_some() && e.network_id != network_id;
!ttl_expired && !released_for_other_networks
});
let answer = match shared {
Some(e) if e.network_id != network_id => codec::EvaluateEnterAnswer::AskAgain,
Some(e) if e.evaluate_enter_id == evaluate_enter_id => {
codec::EvaluateEnterAnswer::Accepted { chosen_lvl: level }
}
Some(_) => codec::EvaluateEnterAnswer::IgnoreNetwork,
None => {
mem.elections.insert(
level,
codec::ElectionRecord {
network_id,
evaluate_enter_id,
granted_at_millis: now_ms,
completed_at_millis: None,
},
);
service
.set_hooking_memory_locally(top, Some(codec::encode_hooking_memory(&mem)))
.await;
codec::EvaluateEnterAnswer::Accepted { chosen_lvl: level }
}
};
if matches!(answer, codec::EvaluateEnterAnswer::Accepted { .. }) {
st.elected = Some(Election {
network_id,
evaluate_enter_id,
granted_at: Instant::now(),
completed_at: None,
});
}
answer
}
async fn release(
&self,
service: &ntk_coordinator::CoordinatorService,
top: usize,
level: usize,
) {
{
let mut levels = self.levels.lock().await;
levels.entry(level).or_default().elected = None;
}
if let Some(tv) = service.hooking_memory_locally(top).await
&& let Ok(mut mem) = codec::decode_hooking_memory(&tv)
&& mem.elections.remove(&level).is_some()
{
service
.set_hooking_memory_locally(top, Some(codec::encode_hooking_memory(&mem)))
.await;
}
}
async fn complete(
&self,
service: &ntk_coordinator::CoordinatorService,
top: usize,
level: usize,
) {
{
let mut levels = self.levels.lock().await;
if let Some(e) = levels.entry(level).or_default().elected.as_mut() {
e.completed_at.get_or_insert_with(Instant::now);
}
}
if let Some(tv) = service.hooking_memory_locally(top).await
&& let Ok(mut mem) = codec::decode_hooking_memory(&tv)
&& let Some(e) = mem.elections.get_mut(&level)
&& e.completed_at_millis.is_none()
{
e.completed_at_millis = Some(codec::now_millis());
service
.set_hooking_memory_locally(top, Some(codec::encode_hooking_memory(&mem)))
.await;
}
}
}
#[derive(Debug)]
pub struct EnterHandlersAdapter {
pub arbiter: Arc<EnterArbiter>,
pub qspn: ntk_qspn::QspnHandle,
pub net: Arc<NetworkInfo>,
pub coordinator_service:
tokio::sync::watch::Receiver<Option<Arc<ntk_coordinator::CoordinatorService>>>,
}
async fn wait_for_coordinator_service(
rx: &tokio::sync::watch::Receiver<Option<Arc<ntk_coordinator::CoordinatorService>>>,
) -> Arc<ntk_coordinator::CoordinatorService> {
let mut rx = rx.clone();
rx.wait_for(Option::is_some)
.await
.expect("coordinator service sender is never dropped before the daemon shuts down")
.clone()
.expect("wait_for guarantees Some")
}
impl EnterHandlersAdapter {
fn chosen_lvl(&self) -> usize {
chosen_lvl_from_snapshot(
&self.qspn.snapshot(),
&self.net,
self.qspn.my_naddr().topology().levels(),
)
}
}
fn chosen_lvl_from_snapshot(snapshot: &RouteSnapshot, net: &NetworkInfo, levels: usize) -> usize {
let mut max_lvl = 0;
for i in 0..levels.saturating_sub(1) {
let has_real_sibling = snapshot
.levels
.get(i)
.into_iter()
.flatten()
.any(|e| !net.is_foreign(e.destination.level, e.destination.pos));
if has_real_sibling {
max_lvl = i + 1;
}
}
max_lvl
}
impl ntk_coordinator::EvaluateEnterHandler for EnterHandlersAdapter {
fn evaluate_enter(
&self,
_top: usize,
data: TypedValue,
_client_tuple: &[u32],
) -> BoxFuture<'_, TypedValue> {
Box::pin(async move {
let answer = match codec::decode_evaluate_enter_request(&data) {
Ok(req) => {
let level = self.chosen_lvl();
let service = wait_for_coordinator_service(&self.coordinator_service).await;
let top = self.qspn.my_naddr().topology().levels();
let answer = self
.arbiter
.decide(&service, top, level, req.network_id, req.evaluate_enter_id)
.await;
tracing::info!(
level,
network_id = req.network_id,
evaluate_enter_id = req.evaluate_enter_id,
?answer,
"coordinator: evaluate_enter"
);
answer
}
Err(_) => codec::EvaluateEnterAnswer::IgnoreNetwork,
};
codec::encode_evaluate_enter_answer(answer)
})
}
}
impl ntk_coordinator::BeginEnterHandler for EnterHandlersAdapter {
fn begin_enter(
&self,
_top: usize,
_data: TypedValue,
_client_tuple: &[u32],
) -> BoxFuture<'_, TypedValue> {
Box::pin(async move { codec::encode_unit() })
}
}
impl ntk_coordinator::CompletedEnterHandler for EnterHandlersAdapter {
fn completed_enter(
&self,
top: usize,
_data: TypedValue,
_client_tuple: &[u32],
) -> BoxFuture<'_, TypedValue> {
let level = top.saturating_sub(1);
Box::pin(async move {
let service = wait_for_coordinator_service(&self.coordinator_service).await;
let coord_top = self.qspn.my_naddr().topology().levels();
self.arbiter.complete(&service, coord_top, level).await;
tracing::info!(level, "coordinator: completed_enter");
codec::encode_unit()
})
}
}
impl ntk_coordinator::AbortEnterHandler for EnterHandlersAdapter {
fn abort_enter(
&self,
top: usize,
_data: TypedValue,
_client_tuple: &[u32],
) -> BoxFuture<'_, TypedValue> {
let level = top.saturating_sub(1);
Box::pin(async move {
let service = wait_for_coordinator_service(&self.coordinator_service).await;
let coord_top = self.qspn.my_naddr().topology().levels();
self.arbiter.release(&service, coord_top, level).await;
codec::encode_unit()
})
}
}
#[derive(Debug)]
pub struct PropagationHandlerAdapter {
pub hooking: tokio::sync::watch::Receiver<Option<HookingHandle>>,
}
async fn wait_for_hooking(
rx: &tokio::sync::watch::Receiver<Option<HookingHandle>>,
) -> HookingHandle {
let mut rx = rx.clone();
rx.wait_for(Option::is_some)
.await
.expect("hooking handle sender is never dropped before the daemon shuts down")
.clone()
.expect("wait_for guarantees Some")
}
impl ntk_coordinator::PropagationHandler for PropagationHandlerAdapter {
fn prepare_migration(&self, _level: usize, data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async move {
let hooking = wait_for_hooking(&self.hooking).await;
if let Ok(migration_id) = codec::decode_migration_id(&data) {
hooking.notify_prepare_migration(migration_id);
}
})
}
fn finish_migration(&self, level: usize, data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async move {
let hooking = wait_for_hooking(&self.hooking).await;
if let Ok(fmd) = codec::decode_finish_migration_data(&data) {
hooking.notify_finish_migration(level, fmd);
}
})
}
fn prepare_enter(&self, _level: usize, data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async move {
let hooking = wait_for_hooking(&self.hooking).await;
if let Ok(enter_id) = codec::decode_enter_id(&data) {
tracing::info!(
enter_id,
"migration-instrumentation: prepare_enter propagation applied locally"
);
hooking.notify_prepare_enter(enter_id);
}
})
}
fn finish_enter(&self, level: usize, data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async move {
let hooking = wait_for_hooking(&self.hooking).await;
if let Ok(fed) = codec::decode_finish_enter_data(&data) {
tracing::info!(
level,
entry_network_id = fed.entry_data.network_id,
entry_pos = ?fed.entry_data.pos,
"migration-instrumentation: finish_enter propagation applied locally"
);
hooking.notify_finish_enter(level, fed);
}
})
}
fn we_have_splitted(&self, level: usize, _data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async move {
tracing::debug!(
level,
"coordinator we_have_splitted propagation: not modeled by this daemon's Hooking wiring, ignored"
);
})
}
}
#[derive(Debug)]
pub struct CoordinatorStubFactoryAdapter {
pub links: Arc<PeerLinks>,
}
impl ntk_coordinator::CoordinatorStubFactory for CoordinatorStubFactoryAdapter {
fn stub_for_each_neighbor(&self) -> Vec<Arc<dyn ntk_coordinator::CoordinatorStub>> {
self.links
.all()
.into_iter()
.map(|(_, client)| {
Arc::new(ntk_coordinator::RpcCoordinatorStub::new(client))
as Arc<dyn ntk_coordinator::CoordinatorStub>
})
.collect()
}
fn stub_for_all_neighbors(&self) -> Arc<dyn ntk_coordinator::CoordinatorStub> {
Arc::new(AllNeighborsCoordinatorStub {
links: self.links.clone(),
})
}
}
struct AllNeighborsCoordinatorStub {
links: Arc<PeerLinks>,
}
macro_rules! fan_out {
($self:ident, $method:ident, $args:ident) => {
Box::pin(async move {
for (_, client) in $self.links.all() {
let stub = ntk_coordinator::RpcCoordinatorStub::new(client);
let _ = stub.$method($args.clone()).await;
}
Ok(())
})
};
}
impl ntk_coordinator::CoordinatorStub for AllNeighborsCoordinatorStub {
fn execute_prepare_migration(
&self,
args: ntk_coordinator::PropagationArgs,
) -> BoxFuture<'_, Result<(), ntk_peerservices::StubCallError>> {
fan_out!(self, execute_prepare_migration, args)
}
fn execute_finish_migration(
&self,
args: ntk_coordinator::PropagationArgs,
) -> BoxFuture<'_, Result<(), ntk_peerservices::StubCallError>> {
fan_out!(self, execute_finish_migration, args)
}
fn execute_prepare_enter(
&self,
args: ntk_coordinator::PropagationArgs,
) -> BoxFuture<'_, Result<(), ntk_peerservices::StubCallError>> {
fan_out!(self, execute_prepare_enter, args)
}
fn execute_finish_enter(
&self,
args: ntk_coordinator::PropagationArgs,
) -> BoxFuture<'_, Result<(), ntk_peerservices::StubCallError>> {
fan_out!(self, execute_finish_enter, args)
}
fn execute_we_have_splitted(
&self,
args: ntk_coordinator::PropagationArgs,
) -> BoxFuture<'_, Result<(), ntk_peerservices::StubCallError>> {
fan_out!(self, execute_we_have_splitted, args)
}
}
#[derive(Debug)]
pub struct RoutingEnvAdapter {
pub qspn: QspnHandle,
pub registry: Arc<LinkRegistry>,
pub links: Arc<PeerLinks>,
}
impl RoutingEnvAdapter {
fn stub_toward(&self, hc: HCoord, skip: Option<QspnArcId>) -> Option<Arc<dyn PeersStub>> {
let snapshot = self.qspn.snapshot();
let link = first_hop_link(&snapshot, &self.registry, hc, skip)?;
let client = self.links.get(link)?;
Some(Arc::new(ntk_peerservices::RpcPeersStub::new(
client,
self.qspn.my_naddr().topology().clone(),
)))
}
fn link_of_stub(&self, stub: &Arc<dyn PeersStub>) -> Option<LinkId> {
let rpc_stub = stub
.as_any()
.downcast_ref::<ntk_peerservices::RpcPeersStub>()?;
self.links
.all()
.into_iter()
.find(|(_, client)| Arc::ptr_eq(client, rpc_stub.client()))
.map(|(id, _)| id)
}
}
impl RoutingEnv for RoutingEnvAdapter {
fn gnode_exists(&self, hc: HCoord) -> bool {
if self.qspn.my_naddr().is_inside(hc).unwrap_or(false) {
return true;
}
self.qspn
.snapshot()
.levels
.get(hc.level)
.is_some_and(|entries| entries.iter().any(|e| e.destination == hc))
}
fn gateway(
&self,
hc: HCoord,
failed: Option<&Arc<dyn PeersStub>>,
) -> Option<Arc<dyn PeersStub>> {
let skip = failed
.and_then(|f| self.link_of_stub(f))
.and_then(|link| self.registry.qspn_arc_of(link));
self.stub_toward(hc, skip)
}
fn dial(&self, n: &TupleNode) -> Option<Arc<dyn PeersStub>> {
let my_naddr = self.qspn.my_naddr();
let target = dial_target(my_naddr, n)?;
let hc = my_naddr.hcoord(&target).ok()??;
self.stub_toward(hc, None)
}
fn nodes_in_my_group(&self, level: usize) -> usize {
estimate_n_nodes(&self.qspn.snapshot(), None, level) as usize
}
fn neighbors(&self) -> Vec<Arc<dyn PeersStub>> {
self.links
.all()
.into_iter()
.map(|(_, client)| {
Arc::new(ntk_peerservices::RpcPeersStub::new(
client,
self.qspn.my_naddr().topology().clone(),
)) as Arc<dyn PeersStub>
})
.collect()
}
}
fn dial_target(my_naddr: &Naddr, n: &TupleNode) -> Option<Naddr> {
let topology = my_naddr.topology();
if n.top() == 0 || n.top() > topology.levels() {
return None;
}
let mut positions = n.positions().to_vec();
positions.extend_from_slice(&my_naddr.positions()[n.top()..]);
Naddr::new(topology.clone(), positions).ok()
}
#[derive(Debug)]
pub struct QspnArcResolverAdapter {
pub registry: Arc<LinkRegistry>,
}
impl ntk_qspn::ArcResolver for QspnArcResolverAdapter {
fn resolve(&self, caller: &ntk_proto::v1::CallerContext) -> Option<QspnArcId> {
let link = self.registry.link_for_caller(caller.src_nic.as_ref()?)?;
self.registry.qspn_arc_of(link)
}
}
#[cfg(test)]
mod estimate_n_nodes_tests {
use super::{NetworkInfo, estimate_n_nodes};
use ntk_common::Cost;
use ntk_qspn::{ArcId, RouteEntry, RoutePath, RouteSnapshot};
fn route_path(nodes_inside: u32) -> RoutePath {
RoutePath {
arc: ArcId::from(0),
hops: Vec::new(),
cost: Cost::Finite(1),
nodes_inside,
}
}
fn entry(level: usize, pos: u32, nodes_inside: u32) -> RouteEntry {
RouteEntry {
destination: ntk_common::HCoord::new(level, pos),
paths: vec![route_path(nodes_inside)],
}
}
#[test]
fn a_solitary_node_counts_only_itself() {
let snapshot = RouteSnapshot {
levels: vec![vec![]],
};
assert_eq!(estimate_n_nodes(&snapshot, None, 0), 1);
}
#[test]
fn level_zero_destinations_always_count_as_one_real_node_each() {
let snapshot = RouteSnapshot {
levels: vec![vec![entry(0, 1, 99), entry(0, 2, 99)]],
};
assert_eq!(estimate_n_nodes(&snapshot, None, 0), 3);
}
#[test]
fn own_level_zero_siblings_are_reflected_in_a_higher_level_count() {
let snapshot = RouteSnapshot {
levels: vec![
vec![entry(0, 1, 1), entry(0, 2, 1)],
vec![entry(1, 1, 3)],
],
};
assert_eq!(estimate_n_nodes(&snapshot, None, 1), 6);
}
#[test]
fn a_foreign_peer_is_excluded_at_every_recursed_level() {
let net = NetworkInfo::new(2, 1);
net.note_foreign(0, 2); net.note_foreign(1, 1); let snapshot = RouteSnapshot {
levels: vec![vec![entry(0, 1, 1), entry(0, 2, 1)], vec![entry(1, 1, 3)]],
};
assert_eq!(estimate_n_nodes(&snapshot, Some(&net), 1), 2);
}
}
#[cfg(test)]
mod dial_target_tests {
use ntk_common::{Naddr, Topology};
use super::{TupleNode, dial_target};
fn topology() -> Topology {
Topology::new([4, 2, 2, 2]).unwrap()
}
#[test]
fn a_single_hop_partial_tuple_widens_to_the_right_full_address() {
let my_naddr = Naddr::new(topology(), vec![2, 1, 0, 1]).unwrap();
let n = TupleNode::new(topology(), vec![3]).unwrap();
let target = dial_target(&my_naddr, &n).expect("a valid top()=1 tuple must resolve");
assert_eq!(target.positions(), &[3, 1, 0, 1]);
}
#[test]
fn a_full_depth_tuple_still_resolves_unchanged() {
let my_naddr = Naddr::new(topology(), vec![2, 1, 0, 1]).unwrap();
let n = TupleNode::new(topology(), vec![3, 1, 0, 1]).unwrap();
let target = dial_target(&my_naddr, &n).expect("a full-depth tuple must resolve");
assert_eq!(target.positions(), &[3, 1, 0, 1]);
}
#[test]
fn an_empty_tuple_is_refused() {
let my_naddr = Naddr::new(topology(), vec![2, 1, 0, 1]).unwrap();
let n = TupleNode::new(topology(), Vec::new()).unwrap();
assert_eq!(dial_target(&my_naddr, &n), None);
}
#[test]
fn a_tuple_naming_more_levels_than_my_topology_has_is_refused() {
let my_naddr = Naddr::new(topology(), vec![2, 1, 0, 1]).unwrap();
let foreign = Topology::new([4, 2, 2, 2, 2]).unwrap();
let n = TupleNode::new(foreign, vec![3, 1, 0, 1, 0]).unwrap();
assert_eq!(dial_target(&my_naddr, &n), None);
}
#[test]
fn a_position_out_of_range_for_my_topology_is_refused() {
let my_naddr = Naddr::new(topology(), vec![2, 1, 0, 1]).unwrap();
let foreign = Topology::new([8, 2, 2, 2]).unwrap();
let n = TupleNode::new(foreign, vec![5]).unwrap();
assert_eq!(dial_target(&my_naddr, &n), None);
}
}
#[cfg(test)]
mod sibling_position_collision_tests {
use super::{NetworkInfo, chosen_lvl_from_snapshot};
use ntk_common::Cost;
use ntk_qspn::{ArcId, RouteEntry, RoutePath, RouteSnapshot};
fn entry(level: usize, pos: u32) -> RouteEntry {
RouteEntry {
destination: ntk_common::HCoord::new(level, pos),
paths: vec![RoutePath {
arc: ArcId::from(0),
hops: Vec::new(),
cost: Cost::Finite(1),
nodes_inside: 1,
}],
}
}
#[test]
fn same_network_confirmation_survives_a_colliding_foreign_report_either_order() {
let confirmed_then_foreign = NetworkInfo::new(1, 1);
confirmed_then_foreign.note_same_network(0, 1);
confirmed_then_foreign.note_foreign(0, 1);
assert!(!confirmed_then_foreign.is_foreign(0, 1));
let foreign_then_confirmed = NetworkInfo::new(1, 1);
foreign_then_confirmed.note_foreign(0, 1);
foreign_then_confirmed.note_same_network(0, 1);
assert!(!foreign_then_confirmed.is_foreign(0, 1));
}
#[test]
fn a_trio_member_with_positions_colliding_a_foreign_trio_still_gets_the_collective_level() {
let net = NetworkInfo::new(2, 91_000_001);
net.note_same_network(0, 0); net.note_same_network(0, 2); net.note_foreign(0, 0); net.note_foreign(0, 1); net.note_foreign(0, 2);
let snapshot = RouteSnapshot {
levels: vec![vec![entry(0, 0), entry(0, 1), entry(0, 2)]],
};
assert_eq!(chosen_lvl_from_snapshot(&snapshot, &net, 2), 1);
}
proptest::proptest! {
#[test]
fn confirmed_positions_never_read_foreign_regardless_of_interleaving(
confirmed in proptest::collection::hash_set(0u32..8, 0..8),
foreign in proptest::collection::hash_set(0u32..8, 0..8),
confirm_first in proptest::bool::ANY,
) {
let net = NetworkInfo::new(1, 1);
if confirm_first {
for &p in &confirmed { net.note_same_network(0, p); }
for &p in &foreign { net.note_foreign(0, p); }
} else {
for &p in &foreign { net.note_foreign(0, p); }
for &p in &confirmed { net.note_same_network(0, p); }
}
for &p in &confirmed {
proptest::prop_assert!(!net.is_foreign(0, p));
}
}
}
}
#[cfg(test)]
mod enter_arbiter_tests {
use std::sync::Arc;
use std::time::Duration;
use futures::future::BoxFuture;
use ntk_common::{HCoord, Naddr, Topology};
use ntk_coordinator::{
AbortEnterHandler, BeginEnterHandler, CompletedEnterHandler, Config as CoordConfig,
CoordinatorMap, CoordinatorService, EnterHandlers, EvaluateEnterHandler,
FakeCoordinatorStubFactory, Manager as CoordinatorManager, PropagationHandler,
};
use ntk_peerservices::{
Config as PeersConfig, Manager as PeersManager, PeersStub, RoutingEnv, TupleNode,
};
use ntk_proto::v1::TypedValue;
use tokio_util::sync::CancellationToken;
use super::{ELECTED_TTL, EnterArbiter};
use crate::node::codec::{self, EvaluateEnterAnswer};
#[derive(Debug, Clone)]
struct TestMap;
impl CoordinatorMap for TestMap {
fn n_nodes(&self) -> u64 {
1
}
fn free_positions(&self, _level: usize) -> Vec<u32> {
vec![0, 1, 2, 3]
}
fn can_reserve(&self, _level: usize) -> bool {
true
}
fn my_pos(&self, _level: usize) -> u32 {
0
}
fn fp_id(&self, _level: usize) -> i64 {
0
}
}
struct NoopHandlers;
impl EvaluateEnterHandler for NoopHandlers {
fn evaluate_enter<'a>(
&'a self,
_top: usize,
data: TypedValue,
_client_tuple: &'a [u32],
) -> BoxFuture<'a, TypedValue> {
Box::pin(async move { data })
}
}
impl BeginEnterHandler for NoopHandlers {
fn begin_enter<'a>(
&'a self,
_top: usize,
data: TypedValue,
_client_tuple: &'a [u32],
) -> BoxFuture<'a, TypedValue> {
Box::pin(async move { data })
}
}
impl CompletedEnterHandler for NoopHandlers {
fn completed_enter<'a>(
&'a self,
_top: usize,
data: TypedValue,
_client_tuple: &'a [u32],
) -> BoxFuture<'a, TypedValue> {
Box::pin(async move { data })
}
}
impl AbortEnterHandler for NoopHandlers {
fn abort_enter<'a>(
&'a self,
_top: usize,
data: TypedValue,
_client_tuple: &'a [u32],
) -> BoxFuture<'a, TypedValue> {
Box::pin(async move { data })
}
}
fn noop_enter_handlers() -> EnterHandlers {
let h = Arc::new(NoopHandlers);
EnterHandlers {
evaluate_enter: h.clone(),
begin_enter: h.clone(),
completed_enter: h.clone(),
abort_enter: h,
}
}
struct NoopPropagationHandler;
impl PropagationHandler for NoopPropagationHandler {
fn prepare_migration(&self, _level: usize, _data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async {})
}
fn finish_migration(&self, _level: usize, _data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async {})
}
fn prepare_enter(&self, _level: usize, _data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async {})
}
fn finish_enter(&self, _level: usize, _data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async {})
}
fn we_have_splitted(&self, _level: usize, _data: TypedValue) -> BoxFuture<'_, ()> {
Box::pin(async {})
}
}
struct SingleNodeEnv;
impl RoutingEnv for SingleNodeEnv {
fn gnode_exists(&self, hc: HCoord) -> bool {
hc.level == 0 && hc.pos == 0
}
fn gateway(
&self,
_hc: HCoord,
_failed: Option<&Arc<dyn PeersStub>>,
) -> Option<Arc<dyn PeersStub>> {
None
}
fn dial(&self, _n: &TupleNode) -> Option<Arc<dyn PeersStub>> {
None
}
fn nodes_in_my_group(&self, _level: usize) -> usize {
1
}
fn neighbors(&self) -> Vec<Arc<dyn PeersStub>> {
Vec::new()
}
}
async fn build_service() -> (Arc<CoordinatorService>, CancellationToken) {
let topology = Topology::new([4]).unwrap();
let my_addr = Naddr::new(topology.clone(), vec![0u32; topology.levels()]).unwrap();
let (peers_manager, peers) = PeersManager::new(
topology.clone(),
my_addr,
Arc::new(SingleNodeEnv),
PeersConfig::default(),
topology.levels(),
);
let cancel = CancellationToken::new();
tokio::spawn(peers_manager.run(cancel.child_token()));
let (coordinator_manager, coordinator) = CoordinatorManager::new(
topology.clone(),
Arc::new(TestMap),
Arc::new(FakeCoordinatorStubFactory::new(Vec::new())),
Arc::new(NoopPropagationHandler),
noop_enter_handlers(),
CoordConfig::default(),
None,
);
tokio::spawn(coordinator_manager.run(cancel.child_token()));
(
Arc::new(CoordinatorService::new(coordinator, peers)),
cancel,
)
}
#[tokio::test]
async fn a_second_concurrent_entrant_for_the_same_network_is_refused_not_reelected() {
let (service, _cancel) = build_service().await;
let arbiter = EnterArbiter::new();
let (level, network_id, top) = (1, 42, 1);
assert_eq!(
arbiter.decide(&service, top, level, network_id, 1).await,
EvaluateEnterAnswer::Accepted { chosen_lvl: level }
);
arbiter.complete(&service, top, level).await;
assert_eq!(
arbiter.decide(&service, top, level, network_id, 2).await,
EvaluateEnterAnswer::IgnoreNetwork
);
assert_eq!(
arbiter.decide(&service, top, level, network_id, 1).await,
EvaluateEnterAnswer::Accepted { chosen_lvl: level }
);
}
#[tokio::test]
async fn a_different_network_is_admitted_immediately_after_the_winner_completes() {
let (service, _cancel) = build_service().await;
let arbiter = EnterArbiter::new();
let (level, top) = (0, 1);
assert_eq!(
arbiter.decide(&service, top, level, 1, 1).await,
EvaluateEnterAnswer::Accepted { chosen_lvl: level },
"the first asker (network 1) wins the election"
);
assert_eq!(
arbiter.decide(&service, top, level, 2, 2).await,
EvaluateEnterAnswer::AskAgain,
"a different network (2) must wait while network 1's episode is in flight"
);
arbiter.complete(&service, top, level).await;
assert_eq!(
arbiter.decide(&service, top, level, 2, 2).await,
EvaluateEnterAnswer::Accepted { chosen_lvl: level },
"both concurrent askers must eventually be accepted: the loser proceeds once the \
winner releases, rather than retrying forever"
);
}
#[tokio::test]
async fn exactly_one_of_any_number_of_concurrent_entrants_is_elected() {
let (service, _cancel) = build_service().await;
let arbiter = EnterArbiter::new();
let (level, network_id, top) = (0, 7, 1);
let mut accepted = 0;
for id in 1..=8 {
if arbiter.decide(&service, top, level, network_id, id).await
== (EvaluateEnterAnswer::Accepted { chosen_lvl: level })
{
accepted += 1;
}
}
assert_eq!(accepted, 1, "exactly one candidate must ever be elected");
}
#[tokio::test]
async fn a_different_target_network_is_told_to_ask_again_not_refused() {
let (service, _cancel) = build_service().await;
let arbiter = EnterArbiter::new();
let (level, top) = (0, 1);
assert_eq!(
arbiter.decide(&service, top, level, 1, 1).await,
EvaluateEnterAnswer::Accepted { chosen_lvl: level }
);
assert_eq!(
arbiter.decide(&service, top, level, 2, 2).await,
EvaluateEnterAnswer::AskAgain
);
}
#[tokio::test]
async fn abort_releases_the_level_for_a_fresh_election() {
let (service, _cancel) = build_service().await;
let arbiter = EnterArbiter::new();
let (level, top) = (0, 1);
assert_eq!(
arbiter.decide(&service, top, level, 1, 1).await,
EvaluateEnterAnswer::Accepted { chosen_lvl: level }
);
arbiter.release(&service, top, level).await;
assert_eq!(
arbiter.decide(&service, top, level, 1, 2).await,
EvaluateEnterAnswer::Accepted { chosen_lvl: level }
);
}
#[tokio::test]
async fn an_expired_replicated_election_self_heals_without_an_abort() {
let (service, _cancel) = build_service().await;
let arbiter = EnterArbiter::new();
let (level, top) = (0, 1);
let stale_millis = codec::now_millis().saturating_sub(
u64::try_from((ELECTED_TTL + Duration::from_millis(1)).as_millis()).unwrap(),
);
let mut mem = codec::HookingMemory::default();
mem.elections.insert(
level,
codec::ElectionRecord {
network_id: 1,
evaluate_enter_id: 1,
granted_at_millis: stale_millis,
completed_at_millis: None,
},
);
service
.set_hooking_memory_locally(top, Some(codec::encode_hooking_memory(&mem)))
.await;
assert_eq!(
arbiter.decide(&service, top, level, 1, 2).await,
EvaluateEnterAnswer::Accepted { chosen_lvl: level }
);
}
#[tokio::test]
async fn two_members_sharing_the_coordinators_record_converge_on_one_election() {
let (service, _cancel) = build_service().await;
let member_a = EnterArbiter::new();
let member_b = EnterArbiter::new();
let (level, network_id, top) = (1, 99, 1);
let answer_a = member_a.decide(&service, top, level, network_id, 1).await;
let answer_b = member_b.decide(&service, top, level, network_id, 2).await;
assert_eq!(
answer_a,
EvaluateEnterAnswer::Accepted { chosen_lvl: level }
);
assert_eq!(answer_b, EvaluateEnterAnswer::IgnoreNetwork);
}
#[tokio::test]
async fn two_members_with_independent_records_would_each_elect_a_different_member() {
let (service_a, _cancel_a) = build_service().await;
let (service_b, _cancel_b) = build_service().await;
let member_a = EnterArbiter::new();
let member_b = EnterArbiter::new();
let (level, network_id, top) = (1, 99, 1);
let answer_a = member_a.decide(&service_a, top, level, network_id, 1).await;
let answer_b = member_b.decide(&service_b, top, level, network_id, 2).await;
assert_eq!(
answer_a,
EvaluateEnterAnswer::Accepted { chosen_lvl: level }
);
assert_eq!(
answer_b,
EvaluateEnterAnswer::Accepted { chosen_lvl: level },
"two independent records reproduce the pre-fix split: both members grant their own election"
);
}
}
#[cfg(test)]
mod coordinator_map_fp_id_tests {
use super::NetworkInfo;
#[test]
fn two_identities_sharing_a_network_id_compute_the_same_fp_id() {
let a = NetworkInfo::new(2, 91_000_001);
let b = NetworkInfo::new(2, 91_000_001);
assert_eq!(a.network_id(), b.network_id());
}
#[test]
fn two_identities_with_different_network_ids_diverge() {
let a = NetworkInfo::new(2, 91_000_001);
let b = NetworkInfo::new(2, 91_000_002);
assert_ne!(a.network_id(), b.network_id());
}
#[test]
fn set_network_id_is_reflected_immediately() {
let a = NetworkInfo::new(2, 91_000_001);
let before = a.network_id();
a.set_network_id(91_000_002);
assert_ne!(before, a.network_id());
assert_eq!(a.network_id(), 91_000_002);
}
}
#[cfg(test)]
mod generation_reset_tests {
use super::{Live, NetworkInfo};
#[test]
fn resetting_for_a_new_generation_clears_the_latch_and_position_facts() {
let net = NetworkInfo::new(1, 7);
net.set_bootstrapped();
net.note_foreign(0, 3);
net.note_same_network(0, 4);
net.reset_for_new_generation();
assert!(!net.is_bootstrapped());
assert!(!net.is_foreign(0, 3));
assert!(net.foreign_positions().is_empty());
assert_eq!(net.network_id(), 7, "the network id is owned by migrate");
}
#[test]
fn a_live_value_swapped_through_one_clone_is_seen_by_every_holder() {
let a = Live::new(1u32);
let b = a.clone();
a.set(2);
assert_eq!(b.get(), 2);
}
}
#[cfg(test)]
mod eldership_cache_tests {
use std::sync::Arc;
use ntk_common::{Cost, Fingerprint, HCoord, Naddr, Topology};
use ntk_qspn::{ArcId, Hop, QspnEvent, RoutePath};
use tokio_util::sync::CancellationToken;
use super::{EldershipCache, RefreshPlan, refresh_eldership};
fn path_ending_at(destination: HCoord) -> RoutePath {
RoutePath {
arc: ArcId::from(1),
hops: vec![
Hop {
arc: ArcId::from(1),
coord: HCoord::new(0, 9),
},
Hop {
arc: ArcId::from(2),
coord: destination,
},
],
cost: Cost::Finite(1),
nodes_inside: 1,
}
}
fn plan_for(events: &[QspnEvent]) -> RefreshPlan {
let mut plan = RefreshPlan::default();
for event in events {
plan.absorb(event);
}
plan
}
#[test]
fn a_path_event_names_only_the_destination_its_hops_end_at() {
let destination = HCoord::new(1, 3);
for event in [
QspnEvent::PathAdded(path_ending_at(destination)),
QspnEvent::PathChanged(path_ending_at(destination)),
QspnEvent::PathRemoved(path_ending_at(destination)),
] {
let plan = plan_for(&[event]);
assert!(!plan.full);
assert_eq!(plan.destinations, [destination].into());
}
}
#[test]
fn destination_events_name_their_destination() {
let plan = plan_for(&[
QspnEvent::DestinationAdded(HCoord::new(0, 1)),
QspnEvent::DestinationRemoved(HCoord::new(0, 2)),
QspnEvent::DestinationAdded(HCoord::new(0, 1)),
]);
assert!(!plan.full);
assert_eq!(
plan.destinations,
[HCoord::new(0, 1), HCoord::new(0, 2)].into()
);
}
#[test]
fn events_that_do_not_name_their_destinations_force_a_full_refresh() {
for event in [
QspnEvent::BootstrapComplete,
QspnEvent::PresenceNotified,
QspnEvent::ChangedFingerprint(1),
QspnEvent::ArcRemoved {
arc: ArcId::from(1),
bad_link: false,
},
] {
assert!(plan_for(&[event]).full);
}
let no_hops = RoutePath {
hops: Vec::new(),
..path_ending_at(HCoord::new(0, 0))
};
assert!(plan_for(&[QspnEvent::PathAdded(no_hops)]).full);
}
#[test]
fn a_nodes_inside_change_requires_no_requery() {
assert!(plan_for(&[QspnEvent::ChangedNodesInside(1)]).is_noop());
}
#[tokio::test]
async fn queued_events_fold_into_one_plan_and_a_lag_forces_a_full_refresh() {
let (tx, mut rx) = tokio::sync::broadcast::channel(2);
tx.send(QspnEvent::DestinationAdded(HCoord::new(0, 1)))
.unwrap();
tx.send(QspnEvent::DestinationAdded(HCoord::new(0, 2)))
.unwrap();
let mut plan = RefreshPlan::default();
plan.absorb_pending(&mut rx);
assert!(!plan.full);
assert_eq!(
plan.destinations,
[HCoord::new(0, 1), HCoord::new(0, 2)].into()
);
for pos in 0..4 {
tx.send(QspnEvent::DestinationAdded(HCoord::new(0, pos)))
.unwrap();
}
let mut plan = RefreshPlan::default();
plan.absorb_pending(&mut rx);
assert!(
plan.full,
"dropped events cannot be attributed to a destination"
);
}
#[tokio::test]
async fn refresh_evicts_destinations_that_are_no_longer_exported() {
let topology = Topology::new([4u32]).unwrap();
let naddr = Naddr::new(topology, vec![0]).unwrap();
let cancel = CancellationToken::new();
let (qspn, _join) = ntk_qspn::spawn(
naddr,
Fingerprint::new(vec![1u8], 0, vec![0]),
ntk_qspn::QspnConfig::default(),
Arc::new(ntk_qspn::FakeQspnStubFactory::new()),
Arc::new(ntk_qspn::FixedThreshold::default()),
Arc::new(ntk_qspn::DefaultArcIdSource::default()),
cancel.clone(),
);
let cache = EldershipCache::default();
cache.set_foreign(0, 1, Some(Some(7)));
cache.set_foreign(0, 2, Some(Some(8)));
let targeted = RefreshPlan {
full: false,
destinations: [HCoord::new(0, 1)].into(),
};
refresh_eldership(&qspn, 1, &cache, &targeted).await;
assert_eq!(
cache.foreign(0, 1),
i32::MAX,
"the named destination is evicted"
);
assert_eq!(
cache.foreign(0, 2),
8,
"an unnamed destination is left alone"
);
assert_ne!(
cache.my(0),
i32::MAX,
"own eldership is read on every refresh"
);
refresh_eldership(&qspn, 1, &cache, &RefreshPlan::everything()).await;
assert_eq!(
cache.foreign(0, 2),
i32::MAX,
"a full refresh evicts every stale entry"
);
cancel.cancel();
}
}