use std::collections::{BTreeMap, BTreeSet};
use crate::route_control::{
RouteAdvertisement, RouteDelta, RouteSnapshot, RouteWithdrawal, MAX_ROUTES_PER_UPDATE,
MAX_ROUTE_IDENTIFIER_LEN, MAX_ROUTE_PATH, MAX_SUBJECT_LEN,
};
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 subject \"{0}\"")]
NoRoute(String),
#[error("invalid advertisement for \"{subject}\": {reason}")]
InvalidAdvertisement { subject: 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,
local: BTreeSet<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(),
local: BTreeSet::new(),
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 register_local(&mut self, subject: &str) -> bool {
let inserted = self.local.insert(subject.to_string());
if inserted {
self.reselect(subject);
}
inserted
}
pub fn resolve(&self, subject: &str) -> Resolution {
if self.local.contains(subject) {
return Resolution::Local;
}
match self.selected.get(subject) {
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 unregister_local(&mut self, subject: &str) -> bool {
let removed = self.local.remove(subject);
if removed {
self.reselect(subject);
}
removed
}
pub(crate) fn local_subjects(&self) -> impl Iterator<Item = &str> {
self.local.iter().map(String::as_str)
}
pub fn reachable_names(&self) -> Vec<String> {
let mut names: BTreeSet<&str> = self.local.iter().map(String::as_str).collect();
names.extend(self.selected.iter().filter_map(|(subject, selected)| {
matches!(selected, SelectedRoute::Route(_)).then_some(subject.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 {
subject: advertisement.subject.clone(),
reason: reason.to_string(),
};
if advertisement.subject.is_empty() {
return Err(invalid("empty subject"));
}
if advertisement.subject.len() > MAX_SUBJECT_LEN {
return Err(invalid("subject exceeds the subject length limit"));
}
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 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.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 hop.is_empty() || hop.len() > MAX_ROUTE_IDENTIFIER_LEN {
return Err(invalid(
"path node identifier is empty 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 subjects = BTreeSet::new();
for advertisement in &snapshot.routes {
self.validate_advertisement(advertiser, advertisement)?;
if !subjects.insert(advertisement.subject.as_str()) {
return Err(RouteError::InvalidAdvertisement {
subject: advertisement.subject.clone(),
reason: "duplicate subject in one snapshot".to_string(),
});
}
}
let mut touched: BTreeSet<String> = self
.by_session
.remove(session)
.unwrap_or_default()
.into_iter()
.collect();
for subject in &touched {
if let Some(per_session) = self.candidates.get_mut(subject) {
per_session.remove(session);
if per_session.is_empty() {
self.candidates.remove(subject);
}
}
}
let mut owned = BTreeSet::new();
for advertisement in &snapshot.routes {
touched.insert(advertisement.subject.clone());
owned.insert(advertisement.subject.clone());
self.candidates
.entry(advertisement.subject.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 subjects = BTreeSet::new();
for advertisement in &delta.upsert {
self.validate_advertisement(advertiser, advertisement)?;
if !subjects.insert(advertisement.subject.as_str()) {
return Err(RouteError::InvalidAdvertisement {
subject: advertisement.subject.clone(),
reason: "duplicate subject in one delta".to_string(),
});
}
}
for withdrawal in &delta.withdraw {
if withdrawal.subject.is_empty() || withdrawal.subject.len() > MAX_SUBJECT_LEN {
return Err(RouteError::InvalidAdvertisement {
subject: withdrawal.subject.clone(),
reason: "withdrawal subject is empty or exceeds the subject 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 {
subject: withdrawal.subject.clone(),
reason: "withdrawal owner identity is missing or exceeds its length limit"
.into(),
});
}
if !subjects.insert(withdrawal.subject.as_str()) {
return Err(RouteError::InvalidAdvertisement {
subject: withdrawal.subject.clone(),
reason: "a subject appears in both upsert and withdraw".to_string(),
});
}
}
let mut touched = BTreeSet::new();
for advertisement in &delta.upsert {
touched.insert(advertisement.subject.clone());
self.candidates
.entry(advertisement.subject.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.subject.clone());
}
for withdrawal in &delta.withdraw {
let Some(per_session) = self.candidates.get_mut(&withdrawal.subject) 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.subject.clone());
if per_session.is_empty() {
self.candidates.remove(&withdrawal.subject);
}
if let Some(owned) = self.by_session.get_mut(session) {
owned.remove(&withdrawal.subject);
}
}
}
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(subjects) = self.by_session.remove(session) else {
return Vec::new();
};
for subject in &subjects {
if let Some(per_session) = self.candidates.get_mut(subject) {
per_session.remove(session);
if per_session.is_empty() {
self.candidates.remove(subject);
}
}
}
self.reselect_all(subjects)
}
pub fn applied_generation(&self, session: &str) -> Option<u64> {
self.session_generation.get(session).copied()
}
fn reselect_all(&mut self, subjects: BTreeSet<String>) -> Vec<String> {
subjects
.into_iter()
.filter(|subject| self.reselect(subject))
.collect()
}
fn reselect(&mut self, subject: &str) -> bool {
let before = self.selected.get(subject).cloned();
let after = self.select(subject);
match after {
Some(selected) => {
let changed = before.as_ref() != Some(&selected);
self.selected.insert(subject.to_string(), selected);
changed
}
None => {
self.selected.remove(subject);
before.is_some()
}
}
}
fn select(&self, subject: &str) -> Option<SelectedRoute> {
if self.local.contains(subject) {
return None;
}
let candidates = self.candidates.get(subject)?;
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.subject) {
Resolution::Route(peer) => peer,
Resolution::Local | Resolution::Unknown | Resolution::Conflicted { .. } => {
return Err(RouteError::NoRoute(envelope.subject.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(subject: &str, hops: Option<u8>) -> Envelope {
Envelope {
v: PROTOCOL_VERSION,
id: "f1".into(),
subject: subject.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(subject: &str, owner: &str, path: &[&str]) -> RouteAdvertisement {
RouteAdvertisement {
subject: subject.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.register_local("chess");
router
.apply_snapshot(
"sess-1",
"leaf-c",
&snapshot(1, vec![advertisement("go", "leaf-c", &["leaf-c"])]),
)
.unwrap();
assert_eq!(router.resolve("chess"), Resolution::Local);
assert_eq!(router.resolve("go"), 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 unregister_local_drops_the_subject_and_resolve_goes_unknown() {
let mut router = RouteTable::new("node-a");
router.register_local("chess");
assert_eq!(router.resolve("chess"), Resolution::Local);
assert!(router.unregister_local("chess"));
assert_eq!(router.resolve("chess"), Resolution::Unknown);
assert!(
!router.unregister_local("chess"),
"unregistering an absent subject reports no change"
);
}
#[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("chess", "owner", &["owner"])]),
)
.unwrap();
let (peer, forwarded) = router.forward(request("chess", 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("chess", "owner", &["owner"])]),
)
.unwrap();
let (_, forwarded) = router.forward(request("chess", 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("chess", "owner", &["owner"])]),
)
.unwrap();
let refused = router.forward(request("chess", 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("chess", 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("chess", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
assert_eq!(router.resolve("chess"), 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("go", "leaf-c", &["leaf-c", "hub"]),
advertisement("chess", "leaf-c", &["leaf-c", "node-a", "hub"]),
],
),
)
.unwrap_err();
assert!(matches!(error, RouteError::InvalidAdvertisement { .. }));
assert_eq!(
router.resolve("go"),
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("chess", "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("chess", "leaf-c", &["leaf-c", first])),
(
second,
advertisement("chess", "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("chess")
};
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("chess", "leaf-c", &["leaf-c", "mid", "aaa"])],
),
)
.unwrap();
router
.apply_snapshot(
"sess-2",
"zzz",
&snapshot(
1,
vec![advertisement("chess", "leaf-c", &["leaf-c", "zzz"])],
),
)
.unwrap();
assert_eq!(router.resolve("chess"), Resolution::Route("zzz".into()));
}
#[test]
fn distinct_owners_conflict_and_multipath_does_not() {
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"hub-a",
&snapshot(
1,
vec![advertisement("chess", "leaf-c", &["leaf-c", "hub-a"])],
),
)
.unwrap();
router
.apply_snapshot(
"sess-2",
"hub-b",
&snapshot(
1,
vec![advertisement("chess", "leaf-c", &["leaf-c", "hub-b"])],
),
)
.unwrap();
assert!(matches!(router.resolve("chess"), Resolution::Route(_)));
router
.apply_snapshot(
"sess-3",
"leaf-d",
&snapshot(1, vec![advertisement("chess", "leaf-d", &["leaf-d"])]),
)
.unwrap();
assert_eq!(
router.resolve("chess"),
Resolution::Conflicted {
owners: vec!["leaf-c".into(), "leaf-d".into()]
}
);
let changed = router.leave("sess-3");
assert_eq!(changed, vec!["chess".to_string()]);
assert!(matches!(router.resolve("chess"), 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("chess", "leaf-c", &["leaf-c", "hub-a"])],
),
)
.unwrap();
router
.apply_snapshot(
"sess-2",
"hub-b",
&snapshot(
1,
vec![advertisement("chess", "leaf-c", &["leaf-c", "hub-b"])],
),
)
.unwrap();
assert_eq!(router.resolve("chess"), Resolution::Route("hub-a".into()));
let changed = router.leave("sess-1");
assert_eq!(changed, vec!["chess".to_string()]);
assert_eq!(
router.resolve("chess"),
Resolution::Route("hub-b".into()),
"the backup path activates without an unknown interval"
);
router.leave("sess-2");
assert_eq!(router.resolve("chess"), 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("chess", "leaf-c", &["leaf-c", "hub"]),
advertisement("go", "leaf-c", &["leaf-c", "hub"]),
],
),
)
.unwrap();
let changed = router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
6,
vec![advertisement("chess", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
assert_eq!(changed, vec!["go".to_string()]);
assert_eq!(router.resolve("go"), Resolution::Unknown);
let stale = router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(4, vec![advertisement("go", "leaf-c", &["leaf-c", "hub"])]),
)
.unwrap_err();
assert!(matches!(stale, RouteError::StaleUpdate { .. }));
let duplicate = router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
6,
vec![advertisement("chess", "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 chess = advertisement("chess", "leaf-c", &["leaf-c", "hub"]);
router
.apply_snapshot("sess-1", "hub", &snapshot(1, vec![chess.clone()]))
.unwrap();
let conflicting = snapshot(1, vec![advertisement("go", "leaf-c", &["leaf-c", "hub"])]);
assert!(matches!(
router.apply_snapshot("sess-1", "hub", &conflicting),
Err(RouteError::GenerationConflict { .. })
));
assert_eq!(router.resolve("chess"), 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("chess", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
router
.apply_delta(
"sess-1",
"hub",
&RouteDelta {
generation: 2,
upsert: Vec::new(),
withdraw: vec![RouteWithdrawal {
subject: "chess".into(),
owner: "leaf-c".into(),
owner_instance: "different-instance".into(),
owner_epoch: 1,
owner_revision: 0,
}],
},
)
.unwrap();
assert_eq!(router.resolve("chess"), 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("chess", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
let add_go = RouteDelta {
generation: 2,
upsert: vec![advertisement("go", "leaf-c", &["leaf-c", "hub"])],
withdraw: Vec::new(),
};
let changed = router.apply_delta("sess-1", "hub", &add_go).unwrap();
assert_eq!(changed, vec!["go".to_string()]);
assert_eq!(router.resolve("go"), Resolution::Route("hub".into()));
assert!(
router
.apply_delta("sess-1", "hub", &add_go)
.unwrap()
.is_empty(),
"an identical generation re-applies idempotently"
);
let stale = RouteDelta {
generation: 1,
upsert: Vec::new(),
withdraw: vec![RouteWithdrawal {
subject: "chess".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("chess"), 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("go"),
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("chess", "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 {
subject: "chess".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("chess"),
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 {
subject: "s".repeat(crate::route_control::MAX_SUBJECT_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("chess", "leaf-c", &["leaf-c", "hub"])],
withdraw: vec![RouteWithdrawal {
subject: "chess".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("chess"), Resolution::Unknown);
}
#[test]
fn a_fresher_owner_incarnation_outranks_a_shorter_stale_path() {
let mut router = RouteTable::new("node-a");
let stale = advertisement("chess", "leaf-c", &["leaf-c", "hub-a"]);
let mut fresh = advertisement("chess", "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("chess"), Resolution::Route("hub-b".into()));
}
#[test]
fn local_ownership_shadows_remote_candidates() {
let mut router = RouteTable::new("node-a");
router
.apply_snapshot(
"sess-1",
"hub",
&snapshot(
1,
vec![advertisement("chess", "leaf-c", &["leaf-c", "hub"])],
),
)
.unwrap();
router.register_local("chess");
assert_eq!(router.resolve("chess"), Resolution::Local);
router.unregister_local("chess");
assert_eq!(
router.resolve("chess"),
Resolution::Route("hub".into()),
"removing the local subject reactivates the transit candidate"
);
}
}