use std::collections::{BTreeMap, BTreeSet};
use crate::route_control::{
validate_node_identifier, RouteAdvertisement, RouteDelta, RouteSnapshot, RouteWithdrawal,
MAX_DESTINATION_LEN, MAX_ROUTES_PER_UPDATE, MAX_ROUTE_IDENTIFIER_LEN, MAX_ROUTE_PATH,
};
use crate::{Envelope, DEFAULT_HOPS};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Resolution {
Local,
Route(String),
Conflicted { owners: Vec<String> },
Unknown,
}
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum RouteError {
#[error("hop limit exceeded")]
HopLimitExceeded,
#[error("no route for destination node \"{0}\"")]
NoRoute(String),
#[error("invalid advertisement for destination \"{destination}\": {reason}")]
InvalidAdvertisement { destination: String, reason: String },
#[error("stale update on session {session}: generation {generation}")]
StaleUpdate { session: String, generation: u64 },
#[error("generation gap on session {session}: got {generation}, expected {expected}")]
GenerationGap {
session: String,
generation: u64,
expected: u64,
},
#[error("route update exceeds limits: {0}")]
LimitExceeded(String),
#[error("conflicting reuse of generation {generation} on session {session}")]
GenerationConflict { session: String, generation: u64 },
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Candidate {
pub advertisement: RouteAdvertisement,
pub session: String,
pub advertiser: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum SelectedRoute {
Route(Candidate),
Conflicted(BTreeSet<String>),
}
#[derive(Clone, PartialEq, Eq)]
enum UpdateIdentity {
Snapshot(Vec<RouteAdvertisement>),
Delta(Vec<RouteAdvertisement>, Vec<RouteWithdrawal>),
}
#[derive(Clone)]
pub(crate) struct RouteTable {
node: String,
candidates: BTreeMap<String, BTreeMap<String, Candidate>>,
by_session: BTreeMap<String, BTreeSet<String>>,
session_generation: BTreeMap<String, u64>,
session_payload: BTreeMap<String, UpdateIdentity>,
selected: BTreeMap<String, SelectedRoute>,
}
impl RouteTable {
pub fn new(node: &str) -> RouteTable {
RouteTable {
node: node.to_string(),
candidates: BTreeMap::new(),
by_session: BTreeMap::new(),
session_generation: BTreeMap::new(),
session_payload: BTreeMap::new(),
selected: BTreeMap::new(),
}
}
pub fn node(&self) -> &str {
&self.node
}
pub fn resolve(&self, destination: &str) -> Resolution {
if destination == self.node {
return Resolution::Local;
}
match self.selected.get(destination) {
Some(SelectedRoute::Route(candidate)) => {
Resolution::Route(candidate.advertiser.clone())
}
Some(SelectedRoute::Conflicted(owners)) => Resolution::Conflicted {
owners: owners.iter().cloned().collect(),
},
None => Resolution::Unknown,
}
}
pub fn reachable_names(&self) -> Vec<String> {
let mut names: BTreeSet<&str> = BTreeSet::from([self.node.as_str()]);
names.extend(self.selected.iter().filter_map(|(destination, selected)| {
matches!(selected, SelectedRoute::Route(_)).then_some(destination.as_str())
}));
names.into_iter().map(str::to_owned).collect()
}
pub(crate) fn validate_advertisement(
&self,
advertiser: &str,
advertisement: &RouteAdvertisement,
) -> Result<(), RouteError> {
let invalid = |reason: &str| RouteError::InvalidAdvertisement {
destination: advertisement.destination.clone(),
reason: reason.to_string(),
};
if advertisement.destination.is_empty() {
return Err(invalid("empty destination node"));
}
if advertisement.destination.len() > MAX_DESTINATION_LEN {
return Err(invalid(
"destination node exceeds the identifier length limit",
));
}
if validate_node_identifier(&advertisement.destination).is_err() {
return Err(invalid("destination node is not a path-safe identifier"));
}
if advertisement.owner.is_empty() || advertisement.owner_instance.is_empty() {
return Err(invalid("missing owner identity"));
}
if advertisement.owner.len() > MAX_ROUTE_IDENTIFIER_LEN
|| advertisement.owner_instance.len() > MAX_ROUTE_IDENTIFIER_LEN
|| advertiser.is_empty()
|| advertiser.len() > MAX_ROUTE_IDENTIFIER_LEN
{
return Err(invalid(
"owner or advertiser identifier exceeds its length limit",
));
}
if validate_node_identifier(&advertisement.owner).is_err()
|| validate_node_identifier(advertiser).is_err()
{
return Err(invalid("owner or advertiser is not a path-safe identifier"));
}
if advertisement.path.is_empty() {
return Err(invalid("empty path"));
}
if advertisement.path.len() > MAX_ROUTE_PATH {
return Err(invalid("path exceeds the route path limit"));
}
if advertisement.path.first().map(String::as_str) != Some(advertisement.owner.as_str()) {
return Err(invalid("path does not begin at the owner"));
}
if advertisement.destination != advertisement.owner {
return Err(invalid("destination node does not match the route owner"));
}
if advertisement.path.last().map(String::as_str) != Some(advertiser) {
return Err(invalid("path does not end at the direct advertiser"));
}
let mut seen = BTreeSet::new();
for hop in &advertisement.path {
if validate_node_identifier(hop).is_err() {
return Err(invalid(
"path node identifier is empty, unsafe, or exceeds its length limit",
));
}
if !seen.insert(hop.as_str()) {
return Err(invalid("path contains a duplicate node"));
}
}
if seen.contains(self.node.as_str()) {
return Err(invalid("path contains the receiving node"));
}
if advertisement.distance as usize != advertisement.path.len() - 1 {
return Err(invalid("distance disagrees with the path length"));
}
Ok(())
}
pub fn apply_snapshot(
&mut self,
session: &str,
advertiser: &str,
snapshot: &RouteSnapshot,
) -> Result<Vec<String>, RouteError> {
let last = self.session_generation.get(session).copied();
let mut canonical = snapshot.routes.clone();
canonical.sort();
if let Some(last) = last {
if snapshot.generation == last {
return if self.session_payload.get(session)
== Some(&UpdateIdentity::Snapshot(canonical.clone()))
{
Ok(Vec::new())
} else {
Err(RouteError::GenerationConflict {
session: session.to_string(),
generation: snapshot.generation,
})
};
}
if snapshot.generation < last {
return Err(RouteError::StaleUpdate {
session: session.to_string(),
generation: snapshot.generation,
});
}
}
if snapshot.routes.len() > MAX_ROUTES_PER_UPDATE {
return Err(RouteError::LimitExceeded(format!(
"{} routes exceed the per-update limit",
snapshot.routes.len()
)));
}
let mut destinations = BTreeSet::new();
for advertisement in &snapshot.routes {
self.validate_advertisement(advertiser, advertisement)?;
if !destinations.insert(advertisement.destination.as_str()) {
return Err(RouteError::InvalidAdvertisement {
destination: advertisement.destination.clone(),
reason: "duplicate destination in one snapshot".to_string(),
});
}
}
let mut touched: BTreeSet<String> = self
.by_session
.remove(session)
.unwrap_or_default()
.into_iter()
.collect();
for destination in &touched {
if let Some(per_session) = self.candidates.get_mut(destination) {
per_session.remove(session);
if per_session.is_empty() {
self.candidates.remove(destination);
}
}
}
let mut owned = BTreeSet::new();
for advertisement in &snapshot.routes {
touched.insert(advertisement.destination.clone());
owned.insert(advertisement.destination.clone());
self.candidates
.entry(advertisement.destination.clone())
.or_default()
.insert(
session.to_string(),
Candidate {
advertisement: advertisement.clone(),
session: session.to_string(),
advertiser: advertiser.to_string(),
},
);
}
if !owned.is_empty() {
self.by_session.insert(session.to_string(), owned);
}
self.session_generation
.insert(session.to_string(), snapshot.generation);
self.session_payload
.insert(session.to_string(), UpdateIdentity::Snapshot(canonical));
Ok(self.reselect_all(touched))
}
pub fn apply_delta(
&mut self,
session: &str,
advertiser: &str,
delta: &RouteDelta,
) -> Result<Vec<String>, RouteError> {
let last = self.session_generation.get(session).copied().unwrap_or(0);
let mut canonical_upsert = delta.upsert.clone();
canonical_upsert.sort();
let mut canonical_withdraw = delta.withdraw.clone();
canonical_withdraw.sort();
if delta.generation == last {
return if self.session_payload.get(session)
== Some(&UpdateIdentity::Delta(
canonical_upsert.clone(),
canonical_withdraw.clone(),
)) {
Ok(Vec::new())
} else {
Err(RouteError::GenerationConflict {
session: session.to_string(),
generation: delta.generation,
})
};
}
if delta.generation < last {
return Err(RouteError::StaleUpdate {
session: session.to_string(),
generation: delta.generation,
});
}
let Some(expected) = last.checked_add(1) else {
return Err(RouteError::LimitExceeded(
"route generation overflow".into(),
));
};
if delta.generation > expected {
return Err(RouteError::GenerationGap {
session: session.to_string(),
generation: delta.generation,
expected,
});
}
let count = delta
.upsert
.len()
.checked_add(delta.withdraw.len())
.ok_or_else(|| RouteError::LimitExceeded("route update count overflow".into()))?;
if count > MAX_ROUTES_PER_UPDATE {
return Err(RouteError::LimitExceeded(format!(
"{} entries exceed the per-update limit",
count
)));
}
let mut destinations = BTreeSet::new();
for advertisement in &delta.upsert {
self.validate_advertisement(advertiser, advertisement)?;
if !destinations.insert(advertisement.destination.as_str()) {
return Err(RouteError::InvalidAdvertisement {
destination: advertisement.destination.clone(),
reason: "duplicate destination in one delta".to_string(),
});
}
}
for withdrawal in &delta.withdraw {
if withdrawal.destination.is_empty()
|| withdrawal.destination.len() > MAX_DESTINATION_LEN
{
return Err(RouteError::InvalidAdvertisement {
destination: withdrawal.destination.clone(),
reason:
"withdrawal destination is empty or exceeds the identifier length limit"
.to_string(),
});
}
if withdrawal.owner.is_empty()
|| withdrawal.owner_instance.is_empty()
|| withdrawal.owner.len() > MAX_ROUTE_IDENTIFIER_LEN
|| withdrawal.owner_instance.len() > MAX_ROUTE_IDENTIFIER_LEN
{
return Err(RouteError::InvalidAdvertisement {
destination: withdrawal.destination.clone(),
reason: "withdrawal owner identity is missing or exceeds its length limit"
.into(),
});
}
if withdrawal.destination != withdrawal.owner {
return Err(RouteError::InvalidAdvertisement {
destination: withdrawal.destination.clone(),
reason: "withdrawal destination does not match the route owner".to_string(),
});
}
if !destinations.insert(withdrawal.destination.as_str()) {
return Err(RouteError::InvalidAdvertisement {
destination: withdrawal.destination.clone(),
reason: "a destination appears in both upsert and withdraw".to_string(),
});
}
}
let mut touched = BTreeSet::new();
for advertisement in &delta.upsert {
touched.insert(advertisement.destination.clone());
self.candidates
.entry(advertisement.destination.clone())
.or_default()
.insert(
session.to_string(),
Candidate {
advertisement: advertisement.clone(),
session: session.to_string(),
advertiser: advertiser.to_string(),
},
);
self.by_session
.entry(session.to_string())
.or_default()
.insert(advertisement.destination.clone());
}
for withdrawal in &delta.withdraw {
let Some(per_session) = self.candidates.get_mut(&withdrawal.destination) else {
continue;
};
let stale = per_session.get(session).is_some_and(|candidate| {
let advertisement = &candidate.advertisement;
(advertisement.owner_epoch, advertisement.owner_revision)
> (withdrawal.owner_epoch, withdrawal.owner_revision)
|| advertisement.owner != withdrawal.owner
|| advertisement.owner_instance != withdrawal.owner_instance
});
if stale {
continue;
}
if per_session.remove(session).is_some() {
touched.insert(withdrawal.destination.clone());
if per_session.is_empty() {
self.candidates.remove(&withdrawal.destination);
}
if let Some(owned) = self.by_session.get_mut(session) {
owned.remove(&withdrawal.destination);
}
}
}
self.session_generation
.insert(session.to_string(), delta.generation);
self.session_payload.insert(
session.to_string(),
UpdateIdentity::Delta(canonical_upsert, canonical_withdraw),
);
Ok(self.reselect_all(touched))
}
pub fn leave(&mut self, session: &str) -> Vec<String> {
self.session_generation.remove(session);
self.session_payload.remove(session);
let Some(destinations) = self.by_session.remove(session) else {
return Vec::new();
};
for destination in &destinations {
if let Some(per_session) = self.candidates.get_mut(destination) {
per_session.remove(session);
if per_session.is_empty() {
self.candidates.remove(destination);
}
}
}
self.reselect_all(destinations)
}
pub fn applied_generation(&self, session: &str) -> Option<u64> {
self.session_generation.get(session).copied()
}
fn reselect_all(&mut self, destinations: BTreeSet<String>) -> Vec<String> {
destinations
.into_iter()
.filter(|destination| self.reselect(destination))
.collect()
}
fn reselect(&mut self, destination: &str) -> bool {
let before = self.selected.get(destination).cloned();
let after = self.select(destination);
match after {
Some(selected) => {
let changed = before.as_ref() != Some(&selected);
self.selected.insert(destination.to_string(), selected);
changed
}
None => {
self.selected.remove(destination);
before.is_some()
}
}
}
fn select(&self, destination: &str) -> Option<SelectedRoute> {
if destination == self.node {
return None;
}
let candidates = self.candidates.get(destination)?;
let newest_epoch: BTreeMap<&str, u64> =
candidates
.values()
.fold(BTreeMap::new(), |mut epochs, candidate| {
let advertisement = &candidate.advertisement;
epochs
.entry(advertisement.owner.as_str())
.and_modify(|epoch| *epoch = (*epoch).max(advertisement.owner_epoch))
.or_insert(advertisement.owner_epoch);
epochs
});
let newest_revision: BTreeMap<(&str, &str, u64), u64> = candidates
.values()
.filter(|candidate| {
let advertisement = &candidate.advertisement;
newest_epoch.get(advertisement.owner.as_str()) == Some(&advertisement.owner_epoch)
})
.fold(BTreeMap::new(), |mut revisions, candidate| {
let advertisement = &candidate.advertisement;
revisions
.entry((
advertisement.owner.as_str(),
advertisement.owner_instance.as_str(),
advertisement.owner_epoch,
))
.and_modify(|revision| {
*revision = (*revision).max(advertisement.owner_revision)
})
.or_insert(advertisement.owner_revision);
revisions
});
let live: Vec<&Candidate> = candidates
.values()
.filter(|candidate| {
let advertisement = &candidate.advertisement;
newest_revision.get(&(
advertisement.owner.as_str(),
advertisement.owner_instance.as_str(),
advertisement.owner_epoch,
)) == Some(&advertisement.owner_revision)
})
.collect();
let owners: BTreeSet<&str> = live
.iter()
.map(|candidate| candidate.advertisement.owner.as_str())
.collect();
if owners.len() > 1 {
return Some(SelectedRoute::Conflicted(
owners.into_iter().map(str::to_owned).collect(),
));
}
let owner = owners.into_iter().next()?;
let incarnations: BTreeSet<&str> = live
.iter()
.filter(|candidate| candidate.advertisement.owner == owner)
.map(|candidate| candidate.advertisement.owner_instance.as_str())
.collect();
if incarnations.len() > 1 {
return Some(SelectedRoute::Conflicted(BTreeSet::from([
owner.to_string()
])));
}
let winner = live.into_iter().min_by(|a, b| {
(
a.advertisement.distance,
&a.advertisement.path,
&a.advertiser,
&a.session,
)
.cmp(&(
b.advertisement.distance,
&b.advertisement.path,
&b.advertiser,
&b.session,
))
})?;
Some(SelectedRoute::Route(winner.clone()))
}
pub(crate) fn selected_transit(&self) -> impl Iterator<Item = &Candidate> {
self.selected
.values()
.filter_map(|selected| match selected {
SelectedRoute::Route(candidate) => Some(candidate),
SelectedRoute::Conflicted(_) => None,
})
}
pub fn forward(&self, mut envelope: Envelope) -> Result<(String, Envelope), RouteError> {
let peer = match self.resolve(&envelope.target) {
Resolution::Route(peer) => peer,
Resolution::Local | Resolution::Unknown | Resolution::Conflicted { .. } => {
return Err(RouteError::NoRoute(envelope.target.clone()))
}
};
let hops = envelope.hops.unwrap_or(DEFAULT_HOPS);
if hops == 0 {
return Err(RouteError::HopLimitExceeded);
}
envelope.hops = Some(hops - 1);
Ok((peer, envelope))
}
pub fn annotate_error(&self, mut envelope: Envelope) -> Envelope {
envelope.path.push(self.node.clone());
envelope
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{Kind, PROTOCOL_VERSION};
use bytes::Bytes;
fn request(destination: &str, hops: Option<u8>) -> Envelope {
Envelope {
v: PROTOCOL_VERSION,
id: "f1".into(),
target: destination.into(),
subject: "service".into(),
kind: Kind::Request,
corr: Some("s1".into()),
seq: None,
hops,
body_token: None,
payload: Bytes::new(),
path: Vec::new(),
headers: Default::default(),
}
}
fn advertisement(destination: &str, owner: &str, path: &[&str]) -> RouteAdvertisement {
RouteAdvertisement {
destination: destination.into(),
owner: owner.into(),
owner_instance: format!("{owner}-inst"),
owner_epoch: 1,
owner_revision: 0,
distance: (path.len() - 1) as u32,
path: path.iter().map(|s| s.to_string()).collect(),
}
}
fn snapshot(generation: u64, routes: Vec<RouteAdvertisement>) -> RouteSnapshot {
RouteSnapshot::canonical(generation, routes)
}
#[test]
fn resolution_order_is_local_then_selected() {
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"leaf-c",
&snapshot(1, vec![advertisement("leaf-c", "leaf-c", &["leaf-c"])]),
)
.unwrap();
assert_eq!(router.resolve("node-a"), Resolution::Local);
assert_eq!(router.resolve("leaf-c"), Resolution::Route("leaf-c".into()));
assert_eq!(router.resolve("weather"), Resolution::Unknown);
}
#[test]
fn no_route_at_all_is_unknown() {
let router = RouteTable::new("island");
assert_eq!(router.resolve("chess"), Resolution::Unknown);
}
#[test]
fn local_destination_is_implicit_and_feature_names_are_unknown() {
let router = RouteTable::new("node-a");
assert_eq!(router.resolve("node-a"), Resolution::Local);
assert_eq!(router.resolve("chess"), Resolution::Unknown);
}
#[test]
fn forward_decrements_hops_toward_the_resolved_peer() {
let mut router = RouteTable::new("relay");
router
.apply_snapshot(
"sess-1",
"owner",
&snapshot(1, vec![advertisement("owner", "owner", &["owner"])]),
)
.unwrap();
let (peer, forwarded) = router.forward(request("owner", Some(8))).unwrap();
assert_eq!(peer, "owner");
assert_eq!(forwarded.hops, Some(7));
}
#[test]
fn missing_hops_default_before_decrement() {
let mut router = RouteTable::new("relay");
router
.apply_snapshot(
"sess-1",
"owner",
&snapshot(1, vec![advertisement("owner", "owner", &["owner"])]),
)
.unwrap();
let (_, forwarded) = router.forward(request("owner", None)).unwrap();
assert_eq!(forwarded.hops, Some(DEFAULT_HOPS - 1));
}
#[test]
fn exhausted_hops_refuse_to_forward() {
let mut router = RouteTable::new("relay");
router
.apply_snapshot(
"sess-1",
"owner",
&snapshot(1, vec![advertisement("owner", "owner", &["owner"])]),
)
.unwrap();
let refused = router.forward(request("owner", Some(0))).unwrap_err();
assert_eq!(refused, RouteError::HopLimitExceeded);
}
#[test]
fn error_frames_accumulate_the_walked_path() {
let router = RouteTable::new("node-b");
let error = Envelope {
kind: Kind::Error,
path: vec!["node-c".into()],
..request("node-c", None)
};
let annotated = router.annotate_error(error);
assert_eq!(
annotated.path,
vec!["node-c".to_string(), "node-b".to_string()]
);
}
#[test]
fn a_transit_advertisement_preserves_the_owner_and_routes_to_the_advertiser() {
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
}
#[test]
fn a_path_containing_the_receiver_is_rejected_atomically() {
let mut router = RouteTable::new("node-a");
let error = router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
1,
vec![
advertisement("node-d", "node-d", &["node-d", "hub"]),
advertisement("leaf-c", "leaf-c", &["leaf-c", "node-a", "hub"]),
],
),
)
.unwrap_err();
assert!(matches!(error, RouteError::InvalidAdvertisement { .. }));
assert_eq!(
router.resolve("node-d"),
Resolution::Unknown,
"an invalid snapshot installs none of its routes"
);
}
#[test]
fn invalid_paths_are_rejected_with_named_reasons() {
let router = RouteTable::new("node-a");
type Mutation = fn(&mut RouteAdvertisement);
let cases: [(Mutation, &str); 6] = [
(|a| a.path = vec![], "empty path"),
(
|a| a.path = vec!["other".into(), "hub".into()],
"begin at the owner",
),
(
|a| a.path = vec!["leaf-c".into(), "other".into()],
"end at the direct advertiser",
),
(
|a| {
a.path = vec!["leaf-c".into(), "x".into(), "x".into(), "hub".into()];
a.distance = 3;
},
"duplicate node",
),
(|a| a.distance = 5, "distance disagrees"),
(
|a| {
a.path = (0..9).map(|i| format!("n{i}")).collect();
a.path[0] = "leaf-c".into();
a.path[8] = "hub".into();
a.distance = 8;
},
"route path limit",
),
];
for (mutation, reason) in cases {
let mut advertisement = advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"]);
mutation(&mut advertisement);
let error = router
.validate_advertisement("hub", &advertisement)
.unwrap_err();
let RouteError::InvalidAdvertisement { reason: named, .. } = &error else {
panic!("expected InvalidAdvertisement, got {error}");
};
assert!(named.contains(reason), "{named} should mention {reason}");
}
}
#[test]
fn same_owner_selection_is_deterministic_regardless_of_arrival_order() {
let build = |first: &str, second: &str| {
let mut router = RouteTable::new("node-a");
let routes = [
(first, advertisement("leaf-c", "leaf-c", &["leaf-c", first])),
(
second,
advertisement("leaf-c", "leaf-c", &["leaf-c", second]),
),
];
for (index, (advertiser, advert)) in routes.iter().enumerate() {
router
.apply_snapshot(
&format!("sess-{advertiser}"),
advertiser,
&snapshot(index as u64 + 1, vec![advert.clone()]),
)
.unwrap();
}
router.resolve("leaf-c")
};
assert_eq!(build("hub-a", "hub-b"), build("hub-b", "hub-a"));
assert_eq!(build("hub-a", "hub-b"), Resolution::Route("hub-a".into()));
}
#[test]
fn lower_distance_wins_over_smaller_path() {
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"aaa",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "mid", "aaa"])],
),
)
.unwrap();
router
.apply_snapshot(
"sess-2",
"zzz",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "zzz"])],
),
)
.unwrap();
assert_eq!(router.resolve("leaf-c"), Resolution::Route("zzz".into()));
}
#[test]
fn duplicate_incarnations_conflict_and_multipath_does_not() {
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"hub-a",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-a"])],
),
)
.unwrap();
router
.apply_snapshot(
"sess-2",
"hub-b",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-b"])],
),
)
.unwrap();
assert!(matches!(router.resolve("leaf-c"), Resolution::Route(_)));
let mut restarted = advertisement("leaf-c", "leaf-c", &["leaf-c"]);
restarted.owner_instance = "leaf-c-restart".into();
router
.apply_snapshot("sess-3", "leaf-c", &snapshot(1, vec![restarted]))
.unwrap();
assert_eq!(
router.resolve("leaf-c"),
Resolution::Conflicted {
owners: vec!["leaf-c".into()]
}
);
let changed = router.leave("sess-3");
assert_eq!(changed, vec!["leaf-c".to_string()]);
assert!(matches!(router.resolve("leaf-c"), Resolution::Route(_)));
}
#[test]
fn session_loss_removes_its_candidates_and_activates_the_backup_atomically() {
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"hub-a",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-a"])],
),
)
.unwrap();
router
.apply_snapshot(
"sess-2",
"hub-b",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-b"])],
),
)
.unwrap();
assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub-a".into()));
let changed = router.leave("sess-1");
assert_eq!(changed, vec!["leaf-c".to_string()]);
assert_eq!(
router.resolve("leaf-c"),
Resolution::Route("hub-b".into()),
"the backup path activates without an unknown interval"
);
router.leave("sess-2");
assert_eq!(router.resolve("leaf-c"), Resolution::Unknown);
}
#[test]
fn a_newer_generation_snapshot_replaces_the_session_view_and_stale_is_rejected() {
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
5,
vec![
advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"]),
advertisement("leaf-d", "leaf-d", &["leaf-d", "hub"]),
],
),
)
.unwrap();
let changed = router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
6,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
assert_eq!(changed, vec!["leaf-d".to_string()]);
assert_eq!(router.resolve("leaf-d"), Resolution::Unknown);
let stale = router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
4,
vec![advertisement("leaf-d", "leaf-d", &["leaf-d", "hub"])],
),
)
.unwrap_err();
assert!(matches!(stale, RouteError::StaleUpdate { .. }));
let duplicate = router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
6,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
assert!(duplicate.is_empty(), "a duplicate generation is idempotent");
}
#[test]
fn duplicate_generation_requires_the_same_canonical_payload() {
let mut router = RouteTable::new("node-a");
let leaf_c = advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"]);
router
.apply_snapshot("sess-1", "hub", &snapshot(1, vec![leaf_c.clone()]))
.unwrap();
let conflicting = snapshot(
1,
vec![advertisement("leaf-d", "leaf-d", &["leaf-d", "hub"])],
);
assert!(matches!(
router.apply_snapshot("sess-1", "hub", &conflicting),
Err(RouteError::GenerationConflict { .. })
));
assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
}
#[test]
fn withdrawal_must_match_the_owner_instance() {
use crate::route_control::{RouteDelta, RouteWithdrawal};
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
router
.apply_delta(
"sess-1",
"hub",
&RouteDelta {
generation: 2,
upsert: Vec::new(),
withdraw: vec![RouteWithdrawal {
destination: "leaf-c".into(),
owner: "leaf-c".into(),
owner_instance: "different-instance".into(),
owner_epoch: 1,
owner_revision: 0,
}],
},
)
.unwrap();
assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
}
#[test]
fn deltas_apply_sequentially_with_duplicate_stale_and_gap_handling() {
use crate::route_control::{RouteDelta, RouteWithdrawal};
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
let add_leaf_d = RouteDelta {
generation: 2,
upsert: vec![advertisement("leaf-d", "leaf-d", &["leaf-d", "hub"])],
withdraw: Vec::new(),
};
let changed = router.apply_delta("sess-1", "hub", &add_leaf_d).unwrap();
assert_eq!(changed, vec!["leaf-d".to_string()]);
assert_eq!(router.resolve("leaf-d"), Resolution::Route("hub".into()));
assert!(
router
.apply_delta("sess-1", "hub", &add_leaf_d)
.unwrap()
.is_empty(),
"an identical generation re-applies idempotently"
);
let stale = RouteDelta {
generation: 1,
upsert: Vec::new(),
withdraw: vec![RouteWithdrawal {
destination: "leaf-c".into(),
owner: "leaf-c".into(),
owner_instance: "leaf-c-inst".into(),
owner_epoch: 1,
owner_revision: 0,
}],
};
assert!(matches!(
router.apply_delta("sess-1", "hub", &stale),
Err(RouteError::StaleUpdate { .. })
));
assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
let gapped = RouteDelta {
generation: 5,
upsert: Vec::new(),
withdraw: Vec::new(),
};
let Err(RouteError::GenerationGap { expected, .. }) =
router.apply_delta("sess-1", "hub", &gapped)
else {
panic!("expected a generation gap");
};
assert_eq!(expected, 3);
assert_eq!(
router.resolve("leaf-d"),
Resolution::Route("hub".into()),
"a gapped delta is not partially applied"
);
}
#[test]
fn a_stale_withdrawal_cannot_remove_a_newer_owner_incarnation() {
use crate::route_control::{RouteDelta, RouteWithdrawal};
let mut router = RouteTable::new("node-a");
let mut fresh = advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"]);
fresh.owner_epoch = 3;
router
.apply_snapshot("sess-1", "hub", &snapshot(1, vec![fresh]))
.unwrap();
let stale_withdrawal = RouteDelta {
generation: 2,
upsert: Vec::new(),
withdraw: vec![RouteWithdrawal {
destination: "leaf-c".into(),
owner: "leaf-c".into(),
owner_instance: "leaf-c-inst".into(),
owner_epoch: 2,
owner_revision: 9,
}],
};
let changed = router
.apply_delta("sess-1", "hub", &stale_withdrawal)
.unwrap();
assert!(changed.is_empty());
assert_eq!(
router.resolve("leaf-c"),
Resolution::Route("hub".into()),
"the newer incarnation survives a stale withdrawal"
);
}
#[test]
fn an_unbounded_withdrawal_subject_rejects_the_whole_delta() {
use crate::route_control::{RouteDelta, RouteWithdrawal};
let mut router = RouteTable::new("node-a");
router
.apply_snapshot("sess-1", "hub", &snapshot(1, Vec::new()))
.unwrap();
let oversized = RouteDelta {
generation: 2,
upsert: Vec::new(),
withdraw: vec![RouteWithdrawal {
destination: "s".repeat(crate::route_control::MAX_DESTINATION_LEN + 1),
owner: "leaf-c".into(),
owner_instance: "leaf-c-inst".into(),
owner_epoch: 1,
owner_revision: 0,
}],
};
assert!(matches!(
router.apply_delta("sess-1", "hub", &oversized),
Err(RouteError::InvalidAdvertisement { .. })
));
}
#[test]
fn a_subject_in_both_upsert_and_withdraw_rejects_the_whole_delta() {
use crate::route_control::{RouteDelta, RouteWithdrawal};
let mut router = RouteTable::new("node-a");
router
.apply_snapshot("sess-1", "hub", &snapshot(1, Vec::new()))
.unwrap();
let contradictory = RouteDelta {
generation: 2,
upsert: vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
withdraw: vec![RouteWithdrawal {
destination: "leaf-c".into(),
owner: "leaf-c".into(),
owner_instance: "leaf-c-inst".into(),
owner_epoch: 1,
owner_revision: 0,
}],
};
assert!(matches!(
router.apply_delta("sess-1", "hub", &contradictory),
Err(RouteError::InvalidAdvertisement { .. })
));
assert_eq!(router.resolve("leaf-c"), Resolution::Unknown);
}
#[test]
fn a_fresher_owner_incarnation_outranks_a_shorter_stale_path() {
let mut router = RouteTable::new("node-a");
let stale = advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-a"]);
let mut fresh = advertisement("leaf-c", "leaf-c", &["leaf-c", "mid", "hub-b"]);
fresh.owner_epoch = 2;
router
.apply_snapshot("sess-1", "hub-a", &snapshot(1, vec![stale]))
.unwrap();
router
.apply_snapshot("sess-2", "hub-b", &snapshot(1, vec![fresh]))
.unwrap();
assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub-b".into()));
}
#[test]
fn implicit_local_destination_is_not_remote_route_state() {
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
1,
vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
assert_eq!(router.resolve("node-a"), Resolution::Local);
assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
}
}