1use std::collections::{BTreeMap, BTreeSet};
2
3use crate::route_control::{
4 validate_node_identifier, RouteAdvertisement, RouteDelta, RouteSnapshot, RouteWithdrawal,
5 MAX_DESTINATION_LEN, MAX_ROUTES_PER_UPDATE, MAX_ROUTE_IDENTIFIER_LEN, MAX_ROUTE_PATH,
6};
7use crate::{Envelope, DEFAULT_HOPS};
8
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub enum Resolution {
11 Local,
12 Route(String),
13 Conflicted { owners: Vec<String> },
14 Unknown,
15}
16
17#[derive(Debug, thiserror::Error, PartialEq, Eq)]
18pub enum RouteError {
19 #[error("hop limit exceeded")]
20 HopLimitExceeded,
21 #[error("no route for destination node \"{0}\"")]
22 NoRoute(String),
23 #[error("invalid advertisement for destination \"{destination}\": {reason}")]
24 InvalidAdvertisement { destination: String, reason: String },
25 #[error("stale update on session {session}: generation {generation}")]
26 StaleUpdate { session: String, generation: u64 },
27 #[error("generation gap on session {session}: got {generation}, expected {expected}")]
28 GenerationGap {
29 session: String,
30 generation: u64,
31 expected: u64,
32 },
33 #[error("route update exceeds limits: {0}")]
34 LimitExceeded(String),
35 #[error("conflicting reuse of generation {generation} on session {session}")]
36 GenerationConflict { session: String, generation: u64 },
37}
38
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub(crate) struct Candidate {
41 pub advertisement: RouteAdvertisement,
42 pub session: String,
43 pub advertiser: String,
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47enum SelectedRoute {
48 Route(Candidate),
49 Conflicted(BTreeSet<String>),
50}
51
52#[derive(Clone, PartialEq, Eq)]
53enum UpdateIdentity {
54 Snapshot(Vec<RouteAdvertisement>),
55 Delta(Vec<RouteAdvertisement>, Vec<RouteWithdrawal>),
56}
57
58#[derive(Clone)]
59pub(crate) struct RouteTable {
60 node: String,
61 candidates: BTreeMap<String, BTreeMap<String, Candidate>>,
62 by_session: BTreeMap<String, BTreeSet<String>>,
63 session_generation: BTreeMap<String, u64>,
64 session_payload: BTreeMap<String, UpdateIdentity>,
65 selected: BTreeMap<String, SelectedRoute>,
66}
67
68impl RouteTable {
69 pub fn new(node: &str) -> RouteTable {
70 RouteTable {
71 node: node.to_string(),
72 candidates: BTreeMap::new(),
73 by_session: BTreeMap::new(),
74 session_generation: BTreeMap::new(),
75 session_payload: BTreeMap::new(),
76 selected: BTreeMap::new(),
77 }
78 }
79
80 pub fn node(&self) -> &str {
81 &self.node
82 }
83
84 pub fn resolve(&self, destination: &str) -> Resolution {
85 if destination == self.node {
86 return Resolution::Local;
87 }
88 match self.selected.get(destination) {
89 Some(SelectedRoute::Route(candidate)) => {
90 Resolution::Route(candidate.advertiser.clone())
91 }
92 Some(SelectedRoute::Conflicted(owners)) => Resolution::Conflicted {
93 owners: owners.iter().cloned().collect(),
94 },
95 None => Resolution::Unknown,
96 }
97 }
98
99 pub fn reachable_names(&self) -> Vec<String> {
100 let mut names: BTreeSet<&str> = BTreeSet::from([self.node.as_str()]);
101 names.extend(self.selected.iter().filter_map(|(destination, selected)| {
102 matches!(selected, SelectedRoute::Route(_)).then_some(destination.as_str())
103 }));
104 names.into_iter().map(str::to_owned).collect()
105 }
106
107 pub(crate) fn validate_advertisement(
108 &self,
109 advertiser: &str,
110 advertisement: &RouteAdvertisement,
111 ) -> Result<(), RouteError> {
112 let invalid = |reason: &str| RouteError::InvalidAdvertisement {
113 destination: advertisement.destination.clone(),
114 reason: reason.to_string(),
115 };
116 if advertisement.destination.is_empty() {
117 return Err(invalid("empty destination node"));
118 }
119 if advertisement.destination.len() > MAX_DESTINATION_LEN {
120 return Err(invalid(
121 "destination node exceeds the identifier length limit",
122 ));
123 }
124 if validate_node_identifier(&advertisement.destination).is_err() {
125 return Err(invalid("destination node is not a path-safe identifier"));
126 }
127 if advertisement.owner.is_empty() || advertisement.owner_instance.is_empty() {
128 return Err(invalid("missing owner identity"));
129 }
130 if advertisement.owner.len() > MAX_ROUTE_IDENTIFIER_LEN
131 || advertisement.owner_instance.len() > MAX_ROUTE_IDENTIFIER_LEN
132 || advertiser.is_empty()
133 || advertiser.len() > MAX_ROUTE_IDENTIFIER_LEN
134 {
135 return Err(invalid(
136 "owner or advertiser identifier exceeds its length limit",
137 ));
138 }
139 if validate_node_identifier(&advertisement.owner).is_err()
140 || validate_node_identifier(advertiser).is_err()
141 {
142 return Err(invalid("owner or advertiser is not a path-safe identifier"));
143 }
144 if advertisement.path.is_empty() {
145 return Err(invalid("empty path"));
146 }
147 if advertisement.path.len() > MAX_ROUTE_PATH {
148 return Err(invalid("path exceeds the route path limit"));
149 }
150 if advertisement.path.first().map(String::as_str) != Some(advertisement.owner.as_str()) {
151 return Err(invalid("path does not begin at the owner"));
152 }
153 if advertisement.destination != advertisement.owner {
154 return Err(invalid("destination node does not match the route owner"));
155 }
156 if advertisement.path.last().map(String::as_str) != Some(advertiser) {
157 return Err(invalid("path does not end at the direct advertiser"));
158 }
159 let mut seen = BTreeSet::new();
160 for hop in &advertisement.path {
161 if validate_node_identifier(hop).is_err() {
162 return Err(invalid(
163 "path node identifier is empty, unsafe, or exceeds its length limit",
164 ));
165 }
166 if !seen.insert(hop.as_str()) {
167 return Err(invalid("path contains a duplicate node"));
168 }
169 }
170 if seen.contains(self.node.as_str()) {
171 return Err(invalid("path contains the receiving node"));
172 }
173 if advertisement.distance as usize != advertisement.path.len() - 1 {
174 return Err(invalid("distance disagrees with the path length"));
175 }
176 Ok(())
177 }
178
179 pub fn apply_snapshot(
180 &mut self,
181 session: &str,
182 advertiser: &str,
183 snapshot: &RouteSnapshot,
184 ) -> Result<Vec<String>, RouteError> {
185 let last = self.session_generation.get(session).copied();
186 let mut canonical = snapshot.routes.clone();
187 canonical.sort();
188 if let Some(last) = last {
189 if snapshot.generation == last {
190 return if self.session_payload.get(session)
191 == Some(&UpdateIdentity::Snapshot(canonical.clone()))
192 {
193 Ok(Vec::new())
194 } else {
195 Err(RouteError::GenerationConflict {
196 session: session.to_string(),
197 generation: snapshot.generation,
198 })
199 };
200 }
201 if snapshot.generation < last {
202 return Err(RouteError::StaleUpdate {
203 session: session.to_string(),
204 generation: snapshot.generation,
205 });
206 }
207 }
208 if snapshot.routes.len() > MAX_ROUTES_PER_UPDATE {
209 return Err(RouteError::LimitExceeded(format!(
210 "{} routes exceed the per-update limit",
211 snapshot.routes.len()
212 )));
213 }
214 let mut destinations = BTreeSet::new();
215 for advertisement in &snapshot.routes {
216 self.validate_advertisement(advertiser, advertisement)?;
217 if !destinations.insert(advertisement.destination.as_str()) {
218 return Err(RouteError::InvalidAdvertisement {
219 destination: advertisement.destination.clone(),
220 reason: "duplicate destination in one snapshot".to_string(),
221 });
222 }
223 }
224 let mut touched: BTreeSet<String> = self
225 .by_session
226 .remove(session)
227 .unwrap_or_default()
228 .into_iter()
229 .collect();
230 for destination in &touched {
231 if let Some(per_session) = self.candidates.get_mut(destination) {
232 per_session.remove(session);
233 if per_session.is_empty() {
234 self.candidates.remove(destination);
235 }
236 }
237 }
238 let mut owned = BTreeSet::new();
239 for advertisement in &snapshot.routes {
240 touched.insert(advertisement.destination.clone());
241 owned.insert(advertisement.destination.clone());
242 self.candidates
243 .entry(advertisement.destination.clone())
244 .or_default()
245 .insert(
246 session.to_string(),
247 Candidate {
248 advertisement: advertisement.clone(),
249 session: session.to_string(),
250 advertiser: advertiser.to_string(),
251 },
252 );
253 }
254 if !owned.is_empty() {
255 self.by_session.insert(session.to_string(), owned);
256 }
257 self.session_generation
258 .insert(session.to_string(), snapshot.generation);
259 self.session_payload
260 .insert(session.to_string(), UpdateIdentity::Snapshot(canonical));
261 Ok(self.reselect_all(touched))
262 }
263
264 pub fn apply_delta(
265 &mut self,
266 session: &str,
267 advertiser: &str,
268 delta: &RouteDelta,
269 ) -> Result<Vec<String>, RouteError> {
270 let last = self.session_generation.get(session).copied().unwrap_or(0);
271 let mut canonical_upsert = delta.upsert.clone();
272 canonical_upsert.sort();
273 let mut canonical_withdraw = delta.withdraw.clone();
274 canonical_withdraw.sort();
275 if delta.generation == last {
276 return if self.session_payload.get(session)
277 == Some(&UpdateIdentity::Delta(
278 canonical_upsert.clone(),
279 canonical_withdraw.clone(),
280 )) {
281 Ok(Vec::new())
282 } else {
283 Err(RouteError::GenerationConflict {
284 session: session.to_string(),
285 generation: delta.generation,
286 })
287 };
288 }
289 if delta.generation < last {
290 return Err(RouteError::StaleUpdate {
291 session: session.to_string(),
292 generation: delta.generation,
293 });
294 }
295 let Some(expected) = last.checked_add(1) else {
296 return Err(RouteError::LimitExceeded(
297 "route generation overflow".into(),
298 ));
299 };
300 if delta.generation > expected {
301 return Err(RouteError::GenerationGap {
302 session: session.to_string(),
303 generation: delta.generation,
304 expected,
305 });
306 }
307 let count = delta
308 .upsert
309 .len()
310 .checked_add(delta.withdraw.len())
311 .ok_or_else(|| RouteError::LimitExceeded("route update count overflow".into()))?;
312 if count > MAX_ROUTES_PER_UPDATE {
313 return Err(RouteError::LimitExceeded(format!(
314 "{} entries exceed the per-update limit",
315 count
316 )));
317 }
318 let mut destinations = BTreeSet::new();
319 for advertisement in &delta.upsert {
320 self.validate_advertisement(advertiser, advertisement)?;
321 if !destinations.insert(advertisement.destination.as_str()) {
322 return Err(RouteError::InvalidAdvertisement {
323 destination: advertisement.destination.clone(),
324 reason: "duplicate destination in one delta".to_string(),
325 });
326 }
327 }
328 for withdrawal in &delta.withdraw {
329 if withdrawal.destination.is_empty()
330 || withdrawal.destination.len() > MAX_DESTINATION_LEN
331 {
332 return Err(RouteError::InvalidAdvertisement {
333 destination: withdrawal.destination.clone(),
334 reason:
335 "withdrawal destination is empty or exceeds the identifier length limit"
336 .to_string(),
337 });
338 }
339 if withdrawal.owner.is_empty()
340 || withdrawal.owner_instance.is_empty()
341 || withdrawal.owner.len() > MAX_ROUTE_IDENTIFIER_LEN
342 || withdrawal.owner_instance.len() > MAX_ROUTE_IDENTIFIER_LEN
343 {
344 return Err(RouteError::InvalidAdvertisement {
345 destination: withdrawal.destination.clone(),
346 reason: "withdrawal owner identity is missing or exceeds its length limit"
347 .into(),
348 });
349 }
350 if withdrawal.destination != withdrawal.owner {
351 return Err(RouteError::InvalidAdvertisement {
352 destination: withdrawal.destination.clone(),
353 reason: "withdrawal destination does not match the route owner".to_string(),
354 });
355 }
356 if !destinations.insert(withdrawal.destination.as_str()) {
357 return Err(RouteError::InvalidAdvertisement {
358 destination: withdrawal.destination.clone(),
359 reason: "a destination appears in both upsert and withdraw".to_string(),
360 });
361 }
362 }
363 let mut touched = BTreeSet::new();
364 for advertisement in &delta.upsert {
365 touched.insert(advertisement.destination.clone());
366 self.candidates
367 .entry(advertisement.destination.clone())
368 .or_default()
369 .insert(
370 session.to_string(),
371 Candidate {
372 advertisement: advertisement.clone(),
373 session: session.to_string(),
374 advertiser: advertiser.to_string(),
375 },
376 );
377 self.by_session
378 .entry(session.to_string())
379 .or_default()
380 .insert(advertisement.destination.clone());
381 }
382 for withdrawal in &delta.withdraw {
383 let Some(per_session) = self.candidates.get_mut(&withdrawal.destination) else {
384 continue;
385 };
386 let stale = per_session.get(session).is_some_and(|candidate| {
387 let advertisement = &candidate.advertisement;
388 (advertisement.owner_epoch, advertisement.owner_revision)
389 > (withdrawal.owner_epoch, withdrawal.owner_revision)
390 || advertisement.owner != withdrawal.owner
391 || advertisement.owner_instance != withdrawal.owner_instance
392 });
393 if stale {
394 continue;
395 }
396 if per_session.remove(session).is_some() {
397 touched.insert(withdrawal.destination.clone());
398 if per_session.is_empty() {
399 self.candidates.remove(&withdrawal.destination);
400 }
401 if let Some(owned) = self.by_session.get_mut(session) {
402 owned.remove(&withdrawal.destination);
403 }
404 }
405 }
406 self.session_generation
407 .insert(session.to_string(), delta.generation);
408 self.session_payload.insert(
409 session.to_string(),
410 UpdateIdentity::Delta(canonical_upsert, canonical_withdraw),
411 );
412 Ok(self.reselect_all(touched))
413 }
414
415 pub fn leave(&mut self, session: &str) -> Vec<String> {
416 self.session_generation.remove(session);
417 self.session_payload.remove(session);
418 let Some(destinations) = self.by_session.remove(session) else {
419 return Vec::new();
420 };
421 for destination in &destinations {
422 if let Some(per_session) = self.candidates.get_mut(destination) {
423 per_session.remove(session);
424 if per_session.is_empty() {
425 self.candidates.remove(destination);
426 }
427 }
428 }
429 self.reselect_all(destinations)
430 }
431
432 pub fn applied_generation(&self, session: &str) -> Option<u64> {
433 self.session_generation.get(session).copied()
434 }
435
436 fn reselect_all(&mut self, destinations: BTreeSet<String>) -> Vec<String> {
437 destinations
438 .into_iter()
439 .filter(|destination| self.reselect(destination))
440 .collect()
441 }
442
443 fn reselect(&mut self, destination: &str) -> bool {
444 let before = self.selected.get(destination).cloned();
445 let after = self.select(destination);
446 match after {
447 Some(selected) => {
448 let changed = before.as_ref() != Some(&selected);
449 self.selected.insert(destination.to_string(), selected);
450 changed
451 }
452 None => {
453 self.selected.remove(destination);
454 before.is_some()
455 }
456 }
457 }
458
459 fn select(&self, destination: &str) -> Option<SelectedRoute> {
460 if destination == self.node {
461 return None;
462 }
463 let candidates = self.candidates.get(destination)?;
464 let newest_epoch: BTreeMap<&str, u64> =
465 candidates
466 .values()
467 .fold(BTreeMap::new(), |mut epochs, candidate| {
468 let advertisement = &candidate.advertisement;
469 epochs
470 .entry(advertisement.owner.as_str())
471 .and_modify(|epoch| *epoch = (*epoch).max(advertisement.owner_epoch))
472 .or_insert(advertisement.owner_epoch);
473 epochs
474 });
475 let newest_revision: BTreeMap<(&str, &str, u64), u64> = candidates
476 .values()
477 .filter(|candidate| {
478 let advertisement = &candidate.advertisement;
479 newest_epoch.get(advertisement.owner.as_str()) == Some(&advertisement.owner_epoch)
480 })
481 .fold(BTreeMap::new(), |mut revisions, candidate| {
482 let advertisement = &candidate.advertisement;
483 revisions
484 .entry((
485 advertisement.owner.as_str(),
486 advertisement.owner_instance.as_str(),
487 advertisement.owner_epoch,
488 ))
489 .and_modify(|revision| {
490 *revision = (*revision).max(advertisement.owner_revision)
491 })
492 .or_insert(advertisement.owner_revision);
493 revisions
494 });
495 let live: Vec<&Candidate> = candidates
496 .values()
497 .filter(|candidate| {
498 let advertisement = &candidate.advertisement;
499 newest_revision.get(&(
500 advertisement.owner.as_str(),
501 advertisement.owner_instance.as_str(),
502 advertisement.owner_epoch,
503 )) == Some(&advertisement.owner_revision)
504 })
505 .collect();
506 let owners: BTreeSet<&str> = live
507 .iter()
508 .map(|candidate| candidate.advertisement.owner.as_str())
509 .collect();
510 if owners.len() > 1 {
511 return Some(SelectedRoute::Conflicted(
512 owners.into_iter().map(str::to_owned).collect(),
513 ));
514 }
515 let owner = owners.into_iter().next()?;
516 let incarnations: BTreeSet<&str> = live
517 .iter()
518 .filter(|candidate| candidate.advertisement.owner == owner)
519 .map(|candidate| candidate.advertisement.owner_instance.as_str())
520 .collect();
521 if incarnations.len() > 1 {
522 return Some(SelectedRoute::Conflicted(BTreeSet::from([
523 owner.to_string()
524 ])));
525 }
526 let winner = live.into_iter().min_by(|a, b| {
527 (
528 a.advertisement.distance,
529 &a.advertisement.path,
530 &a.advertiser,
531 &a.session,
532 )
533 .cmp(&(
534 b.advertisement.distance,
535 &b.advertisement.path,
536 &b.advertiser,
537 &b.session,
538 ))
539 })?;
540 Some(SelectedRoute::Route(winner.clone()))
541 }
542
543 pub(crate) fn selected_transit(&self) -> impl Iterator<Item = &Candidate> {
544 self.selected
545 .values()
546 .filter_map(|selected| match selected {
547 SelectedRoute::Route(candidate) => Some(candidate),
548 SelectedRoute::Conflicted(_) => None,
549 })
550 }
551
552 pub fn forward(&self, mut envelope: Envelope) -> Result<(String, Envelope), RouteError> {
553 let peer = match self.resolve(&envelope.target) {
554 Resolution::Route(peer) => peer,
555 Resolution::Local | Resolution::Unknown | Resolution::Conflicted { .. } => {
556 return Err(RouteError::NoRoute(envelope.target.clone()))
557 }
558 };
559 let hops = envelope.hops.unwrap_or(DEFAULT_HOPS);
560 if hops == 0 {
561 return Err(RouteError::HopLimitExceeded);
562 }
563 envelope.hops = Some(hops - 1);
564 Ok((peer, envelope))
565 }
566
567 pub fn annotate_error(&self, mut envelope: Envelope) -> Envelope {
568 envelope.path.push(self.node.clone());
569 envelope
570 }
571}
572
573#[cfg(test)]
574mod tests {
575 use super::*;
576 use crate::{Kind, PROTOCOL_VERSION};
577 use bytes::Bytes;
578
579 fn request(destination: &str, hops: Option<u8>) -> Envelope {
580 Envelope {
581 v: PROTOCOL_VERSION,
582 id: "f1".into(),
583 target: destination.into(),
584 subject: "service".into(),
585 kind: Kind::Request,
586 corr: Some("s1".into()),
587 seq: None,
588 hops,
589 body_token: None,
590 payload: Bytes::new(),
591 path: Vec::new(),
592 headers: Default::default(),
593 }
594 }
595
596 fn advertisement(destination: &str, owner: &str, path: &[&str]) -> RouteAdvertisement {
597 RouteAdvertisement {
598 destination: destination.into(),
599 owner: owner.into(),
600 owner_instance: format!("{owner}-inst"),
601 owner_epoch: 1,
602 owner_revision: 0,
603 distance: (path.len() - 1) as u32,
604 path: path.iter().map(|s| s.to_string()).collect(),
605 }
606 }
607
608 fn snapshot(generation: u64, routes: Vec<RouteAdvertisement>) -> RouteSnapshot {
609 RouteSnapshot::canonical(generation, routes)
610 }
611
612 #[test]
613 fn resolution_order_is_local_then_selected() {
614 let mut router = RouteTable::new("node-a");
615 router
616 .apply_snapshot(
617 "sess-1",
618 "leaf-c",
619 &snapshot(1, vec![advertisement("leaf-c", "leaf-c", &["leaf-c"])]),
620 )
621 .unwrap();
622
623 assert_eq!(router.resolve("node-a"), Resolution::Local);
624 assert_eq!(router.resolve("leaf-c"), Resolution::Route("leaf-c".into()));
625 assert_eq!(router.resolve("weather"), Resolution::Unknown);
626 }
627
628 #[test]
629 fn no_route_at_all_is_unknown() {
630 let router = RouteTable::new("island");
631 assert_eq!(router.resolve("chess"), Resolution::Unknown);
632 }
633
634 #[test]
635 fn local_destination_is_implicit_and_feature_names_are_unknown() {
636 let router = RouteTable::new("node-a");
637 assert_eq!(router.resolve("node-a"), Resolution::Local);
638 assert_eq!(router.resolve("chess"), Resolution::Unknown);
639 }
640
641 #[test]
642 fn forward_decrements_hops_toward_the_resolved_peer() {
643 let mut router = RouteTable::new("relay");
644 router
645 .apply_snapshot(
646 "sess-1",
647 "owner",
648 &snapshot(1, vec![advertisement("owner", "owner", &["owner"])]),
649 )
650 .unwrap();
651 let (peer, forwarded) = router.forward(request("owner", Some(8))).unwrap();
652 assert_eq!(peer, "owner");
653 assert_eq!(forwarded.hops, Some(7));
654 }
655
656 #[test]
657 fn missing_hops_default_before_decrement() {
658 let mut router = RouteTable::new("relay");
659 router
660 .apply_snapshot(
661 "sess-1",
662 "owner",
663 &snapshot(1, vec![advertisement("owner", "owner", &["owner"])]),
664 )
665 .unwrap();
666 let (_, forwarded) = router.forward(request("owner", None)).unwrap();
667 assert_eq!(forwarded.hops, Some(DEFAULT_HOPS - 1));
668 }
669
670 #[test]
671 fn exhausted_hops_refuse_to_forward() {
672 let mut router = RouteTable::new("relay");
673 router
674 .apply_snapshot(
675 "sess-1",
676 "owner",
677 &snapshot(1, vec![advertisement("owner", "owner", &["owner"])]),
678 )
679 .unwrap();
680 let refused = router.forward(request("owner", Some(0))).unwrap_err();
681 assert_eq!(refused, RouteError::HopLimitExceeded);
682 }
683
684 #[test]
685 fn error_frames_accumulate_the_walked_path() {
686 let router = RouteTable::new("node-b");
687 let error = Envelope {
688 kind: Kind::Error,
689 path: vec!["node-c".into()],
690 ..request("node-c", None)
691 };
692 let annotated = router.annotate_error(error);
693 assert_eq!(
694 annotated.path,
695 vec!["node-c".to_string(), "node-b".to_string()]
696 );
697 }
698
699 #[test]
700 fn a_transit_advertisement_preserves_the_owner_and_routes_to_the_advertiser() {
701 let mut router = RouteTable::new("node-a");
702 router
703 .apply_snapshot(
704 "sess-1",
705 "hub",
706 &snapshot(
707 1,
708 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
709 ),
710 )
711 .unwrap();
712 assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
713 }
714
715 #[test]
716 fn a_path_containing_the_receiver_is_rejected_atomically() {
717 let mut router = RouteTable::new("node-a");
718 let error = router
719 .apply_snapshot(
720 "sess-1",
721 "hub",
722 &snapshot(
723 1,
724 vec![
725 advertisement("node-d", "node-d", &["node-d", "hub"]),
726 advertisement("leaf-c", "leaf-c", &["leaf-c", "node-a", "hub"]),
727 ],
728 ),
729 )
730 .unwrap_err();
731 assert!(matches!(error, RouteError::InvalidAdvertisement { .. }));
732 assert_eq!(
733 router.resolve("node-d"),
734 Resolution::Unknown,
735 "an invalid snapshot installs none of its routes"
736 );
737 }
738
739 #[test]
740 fn invalid_paths_are_rejected_with_named_reasons() {
741 let router = RouteTable::new("node-a");
742 type Mutation = fn(&mut RouteAdvertisement);
743 let cases: [(Mutation, &str); 6] = [
744 (|a| a.path = vec![], "empty path"),
745 (
746 |a| a.path = vec!["other".into(), "hub".into()],
747 "begin at the owner",
748 ),
749 (
750 |a| a.path = vec!["leaf-c".into(), "other".into()],
751 "end at the direct advertiser",
752 ),
753 (
754 |a| {
755 a.path = vec!["leaf-c".into(), "x".into(), "x".into(), "hub".into()];
756 a.distance = 3;
757 },
758 "duplicate node",
759 ),
760 (|a| a.distance = 5, "distance disagrees"),
761 (
762 |a| {
763 a.path = (0..9).map(|i| format!("n{i}")).collect();
764 a.path[0] = "leaf-c".into();
765 a.path[8] = "hub".into();
766 a.distance = 8;
767 },
768 "route path limit",
769 ),
770 ];
771 for (mutation, reason) in cases {
772 let mut advertisement = advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"]);
773 mutation(&mut advertisement);
774 let error = router
775 .validate_advertisement("hub", &advertisement)
776 .unwrap_err();
777 let RouteError::InvalidAdvertisement { reason: named, .. } = &error else {
778 panic!("expected InvalidAdvertisement, got {error}");
779 };
780 assert!(named.contains(reason), "{named} should mention {reason}");
781 }
782 }
783
784 #[test]
785 fn same_owner_selection_is_deterministic_regardless_of_arrival_order() {
786 let build = |first: &str, second: &str| {
787 let mut router = RouteTable::new("node-a");
788 let routes = [
789 (first, advertisement("leaf-c", "leaf-c", &["leaf-c", first])),
790 (
791 second,
792 advertisement("leaf-c", "leaf-c", &["leaf-c", second]),
793 ),
794 ];
795 for (index, (advertiser, advert)) in routes.iter().enumerate() {
796 router
797 .apply_snapshot(
798 &format!("sess-{advertiser}"),
799 advertiser,
800 &snapshot(index as u64 + 1, vec![advert.clone()]),
801 )
802 .unwrap();
803 }
804 router.resolve("leaf-c")
805 };
806 assert_eq!(build("hub-a", "hub-b"), build("hub-b", "hub-a"));
807 assert_eq!(build("hub-a", "hub-b"), Resolution::Route("hub-a".into()));
808 }
809
810 #[test]
811 fn lower_distance_wins_over_smaller_path() {
812 let mut router = RouteTable::new("node-a");
813 router
814 .apply_snapshot(
815 "sess-1",
816 "aaa",
817 &snapshot(
818 1,
819 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "mid", "aaa"])],
820 ),
821 )
822 .unwrap();
823 router
824 .apply_snapshot(
825 "sess-2",
826 "zzz",
827 &snapshot(
828 1,
829 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "zzz"])],
830 ),
831 )
832 .unwrap();
833 assert_eq!(router.resolve("leaf-c"), Resolution::Route("zzz".into()));
834 }
835
836 #[test]
837 fn duplicate_incarnations_conflict_and_multipath_does_not() {
838 let mut router = RouteTable::new("node-a");
839 router
840 .apply_snapshot(
841 "sess-1",
842 "hub-a",
843 &snapshot(
844 1,
845 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-a"])],
846 ),
847 )
848 .unwrap();
849 router
850 .apply_snapshot(
851 "sess-2",
852 "hub-b",
853 &snapshot(
854 1,
855 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-b"])],
856 ),
857 )
858 .unwrap();
859 assert!(matches!(router.resolve("leaf-c"), Resolution::Route(_)));
860
861 let mut restarted = advertisement("leaf-c", "leaf-c", &["leaf-c"]);
862 restarted.owner_instance = "leaf-c-restart".into();
863 router
864 .apply_snapshot("sess-3", "leaf-c", &snapshot(1, vec![restarted]))
865 .unwrap();
866 assert_eq!(
867 router.resolve("leaf-c"),
868 Resolution::Conflicted {
869 owners: vec!["leaf-c".into()]
870 }
871 );
872
873 let changed = router.leave("sess-3");
874 assert_eq!(changed, vec!["leaf-c".to_string()]);
875 assert!(matches!(router.resolve("leaf-c"), Resolution::Route(_)));
876 }
877
878 #[test]
879 fn session_loss_removes_its_candidates_and_activates_the_backup_atomically() {
880 let mut router = RouteTable::new("node-a");
881 router
882 .apply_snapshot(
883 "sess-1",
884 "hub-a",
885 &snapshot(
886 1,
887 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-a"])],
888 ),
889 )
890 .unwrap();
891 router
892 .apply_snapshot(
893 "sess-2",
894 "hub-b",
895 &snapshot(
896 1,
897 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-b"])],
898 ),
899 )
900 .unwrap();
901 assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub-a".into()));
902
903 let changed = router.leave("sess-1");
904 assert_eq!(changed, vec!["leaf-c".to_string()]);
905 assert_eq!(
906 router.resolve("leaf-c"),
907 Resolution::Route("hub-b".into()),
908 "the backup path activates without an unknown interval"
909 );
910
911 router.leave("sess-2");
912 assert_eq!(router.resolve("leaf-c"), Resolution::Unknown);
913 }
914
915 #[test]
916 fn a_newer_generation_snapshot_replaces_the_session_view_and_stale_is_rejected() {
917 let mut router = RouteTable::new("node-a");
918 router
919 .apply_snapshot(
920 "sess-1",
921 "hub",
922 &snapshot(
923 5,
924 vec![
925 advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"]),
926 advertisement("leaf-d", "leaf-d", &["leaf-d", "hub"]),
927 ],
928 ),
929 )
930 .unwrap();
931 let changed = router
932 .apply_snapshot(
933 "sess-1",
934 "hub",
935 &snapshot(
936 6,
937 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
938 ),
939 )
940 .unwrap();
941 assert_eq!(changed, vec!["leaf-d".to_string()]);
942 assert_eq!(router.resolve("leaf-d"), Resolution::Unknown);
943
944 let stale = router
945 .apply_snapshot(
946 "sess-1",
947 "hub",
948 &snapshot(
949 4,
950 vec![advertisement("leaf-d", "leaf-d", &["leaf-d", "hub"])],
951 ),
952 )
953 .unwrap_err();
954 assert!(matches!(stale, RouteError::StaleUpdate { .. }));
955
956 let duplicate = router
957 .apply_snapshot(
958 "sess-1",
959 "hub",
960 &snapshot(
961 6,
962 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
963 ),
964 )
965 .unwrap();
966 assert!(duplicate.is_empty(), "a duplicate generation is idempotent");
967 }
968
969 #[test]
970 fn duplicate_generation_requires_the_same_canonical_payload() {
971 let mut router = RouteTable::new("node-a");
972 let leaf_c = advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"]);
973 router
974 .apply_snapshot("sess-1", "hub", &snapshot(1, vec![leaf_c.clone()]))
975 .unwrap();
976 let conflicting = snapshot(
977 1,
978 vec![advertisement("leaf-d", "leaf-d", &["leaf-d", "hub"])],
979 );
980 assert!(matches!(
981 router.apply_snapshot("sess-1", "hub", &conflicting),
982 Err(RouteError::GenerationConflict { .. })
983 ));
984 assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
985 }
986
987 #[test]
988 fn withdrawal_must_match_the_owner_instance() {
989 use crate::route_control::{RouteDelta, RouteWithdrawal};
990 let mut router = RouteTable::new("node-a");
991 router
992 .apply_snapshot(
993 "sess-1",
994 "hub",
995 &snapshot(
996 1,
997 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
998 ),
999 )
1000 .unwrap();
1001 router
1002 .apply_delta(
1003 "sess-1",
1004 "hub",
1005 &RouteDelta {
1006 generation: 2,
1007 upsert: Vec::new(),
1008 withdraw: vec![RouteWithdrawal {
1009 destination: "leaf-c".into(),
1010 owner: "leaf-c".into(),
1011 owner_instance: "different-instance".into(),
1012 owner_epoch: 1,
1013 owner_revision: 0,
1014 }],
1015 },
1016 )
1017 .unwrap();
1018 assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
1019 }
1020
1021 #[test]
1022 fn deltas_apply_sequentially_with_duplicate_stale_and_gap_handling() {
1023 use crate::route_control::{RouteDelta, RouteWithdrawal};
1024 let mut router = RouteTable::new("node-a");
1025 router
1026 .apply_snapshot(
1027 "sess-1",
1028 "hub",
1029 &snapshot(
1030 1,
1031 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
1032 ),
1033 )
1034 .unwrap();
1035
1036 let add_leaf_d = RouteDelta {
1037 generation: 2,
1038 upsert: vec![advertisement("leaf-d", "leaf-d", &["leaf-d", "hub"])],
1039 withdraw: Vec::new(),
1040 };
1041 let changed = router.apply_delta("sess-1", "hub", &add_leaf_d).unwrap();
1042 assert_eq!(changed, vec!["leaf-d".to_string()]);
1043 assert_eq!(router.resolve("leaf-d"), Resolution::Route("hub".into()));
1044
1045 assert!(
1046 router
1047 .apply_delta("sess-1", "hub", &add_leaf_d)
1048 .unwrap()
1049 .is_empty(),
1050 "an identical generation re-applies idempotently"
1051 );
1052
1053 let stale = RouteDelta {
1054 generation: 1,
1055 upsert: Vec::new(),
1056 withdraw: vec![RouteWithdrawal {
1057 destination: "leaf-c".into(),
1058 owner: "leaf-c".into(),
1059 owner_instance: "leaf-c-inst".into(),
1060 owner_epoch: 1,
1061 owner_revision: 0,
1062 }],
1063 };
1064 assert!(matches!(
1065 router.apply_delta("sess-1", "hub", &stale),
1066 Err(RouteError::StaleUpdate { .. })
1067 ));
1068 assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
1069
1070 let gapped = RouteDelta {
1071 generation: 5,
1072 upsert: Vec::new(),
1073 withdraw: Vec::new(),
1074 };
1075 let Err(RouteError::GenerationGap { expected, .. }) =
1076 router.apply_delta("sess-1", "hub", &gapped)
1077 else {
1078 panic!("expected a generation gap");
1079 };
1080 assert_eq!(expected, 3);
1081 assert_eq!(
1082 router.resolve("leaf-d"),
1083 Resolution::Route("hub".into()),
1084 "a gapped delta is not partially applied"
1085 );
1086 }
1087
1088 #[test]
1089 fn a_stale_withdrawal_cannot_remove_a_newer_owner_incarnation() {
1090 use crate::route_control::{RouteDelta, RouteWithdrawal};
1091 let mut router = RouteTable::new("node-a");
1092 let mut fresh = advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"]);
1093 fresh.owner_epoch = 3;
1094 router
1095 .apply_snapshot("sess-1", "hub", &snapshot(1, vec![fresh]))
1096 .unwrap();
1097
1098 let stale_withdrawal = RouteDelta {
1099 generation: 2,
1100 upsert: Vec::new(),
1101 withdraw: vec![RouteWithdrawal {
1102 destination: "leaf-c".into(),
1103 owner: "leaf-c".into(),
1104 owner_instance: "leaf-c-inst".into(),
1105 owner_epoch: 2,
1106 owner_revision: 9,
1107 }],
1108 };
1109 let changed = router
1110 .apply_delta("sess-1", "hub", &stale_withdrawal)
1111 .unwrap();
1112 assert!(changed.is_empty());
1113 assert_eq!(
1114 router.resolve("leaf-c"),
1115 Resolution::Route("hub".into()),
1116 "the newer incarnation survives a stale withdrawal"
1117 );
1118 }
1119
1120 #[test]
1121 fn an_unbounded_withdrawal_subject_rejects_the_whole_delta() {
1122 use crate::route_control::{RouteDelta, RouteWithdrawal};
1123 let mut router = RouteTable::new("node-a");
1124 router
1125 .apply_snapshot("sess-1", "hub", &snapshot(1, Vec::new()))
1126 .unwrap();
1127 let oversized = RouteDelta {
1128 generation: 2,
1129 upsert: Vec::new(),
1130 withdraw: vec![RouteWithdrawal {
1131 destination: "s".repeat(crate::route_control::MAX_DESTINATION_LEN + 1),
1132 owner: "leaf-c".into(),
1133 owner_instance: "leaf-c-inst".into(),
1134 owner_epoch: 1,
1135 owner_revision: 0,
1136 }],
1137 };
1138 assert!(matches!(
1139 router.apply_delta("sess-1", "hub", &oversized),
1140 Err(RouteError::InvalidAdvertisement { .. })
1141 ));
1142 }
1143
1144 #[test]
1145 fn a_subject_in_both_upsert_and_withdraw_rejects_the_whole_delta() {
1146 use crate::route_control::{RouteDelta, RouteWithdrawal};
1147 let mut router = RouteTable::new("node-a");
1148 router
1149 .apply_snapshot("sess-1", "hub", &snapshot(1, Vec::new()))
1150 .unwrap();
1151 let contradictory = RouteDelta {
1152 generation: 2,
1153 upsert: vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
1154 withdraw: vec![RouteWithdrawal {
1155 destination: "leaf-c".into(),
1156 owner: "leaf-c".into(),
1157 owner_instance: "leaf-c-inst".into(),
1158 owner_epoch: 1,
1159 owner_revision: 0,
1160 }],
1161 };
1162 assert!(matches!(
1163 router.apply_delta("sess-1", "hub", &contradictory),
1164 Err(RouteError::InvalidAdvertisement { .. })
1165 ));
1166 assert_eq!(router.resolve("leaf-c"), Resolution::Unknown);
1167 }
1168
1169 #[test]
1170 fn a_fresher_owner_incarnation_outranks_a_shorter_stale_path() {
1171 let mut router = RouteTable::new("node-a");
1172 let stale = advertisement("leaf-c", "leaf-c", &["leaf-c", "hub-a"]);
1173 let mut fresh = advertisement("leaf-c", "leaf-c", &["leaf-c", "mid", "hub-b"]);
1174 fresh.owner_epoch = 2;
1175 router
1176 .apply_snapshot("sess-1", "hub-a", &snapshot(1, vec![stale]))
1177 .unwrap();
1178 router
1179 .apply_snapshot("sess-2", "hub-b", &snapshot(1, vec![fresh]))
1180 .unwrap();
1181 assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub-b".into()));
1182 }
1183
1184 #[test]
1185 fn implicit_local_destination_is_not_remote_route_state() {
1186 let mut router = RouteTable::new("node-a");
1187 router
1188 .apply_snapshot(
1189 "sess-1",
1190 "hub",
1191 &snapshot(
1192 1,
1193 vec![advertisement("leaf-c", "leaf-c", &["leaf-c", "hub"])],
1194 ),
1195 )
1196 .unwrap();
1197 assert_eq!(router.resolve("node-a"), Resolution::Local);
1198 assert_eq!(router.resolve("leaf-c"), Resolution::Route("hub".into()));
1199 }
1200}