1use std::{cmp::Ordering, sync::Arc};
2
3use hopr_api::{
4 OffchainPublicKey,
5 graph::{
6 NetworkGraphTraverse, NetworkGraphView,
7 function::{BasicValueFn, EdgeValueFn},
8 traits::{
9 EdgeImmediateProtocolObservable, EdgeLinkObservable, EdgeObservableRead, EdgeProtocolObservable, ValueFn,
10 },
11 },
12 types::internal::errors::PathError,
13};
14
15use super::{
16 errors::{PathPlannerError, Result},
17 traits::{PathSelector, PathWithMetrics},
18};
19
20#[derive(Clone, Debug)]
25struct PathCostWithMetrics {
26 cost: f64,
27 total_latency_ms: Option<u32>,
28 min_probe_success_rate: Option<f64>,
29 min_ack_rate: Option<f64>,
30 fundable_tickets_floor: Option<u128>,
31}
32
33impl PartialOrd for PathCostWithMetrics {
34 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
35 self.cost.partial_cmp(&other.cost)
36 }
37}
38
39impl PartialEq for PathCostWithMetrics {
40 fn eq(&self, other: &Self) -> bool {
41 self.cost == other.cost
42 }
43}
44
45impl From<(PathCostWithMetrics, Vec<OffchainPublicKey>)> for PathWithMetrics {
46 fn from((metrics, path): (PathCostWithMetrics, Vec<OffchainPublicKey>)) -> Self {
47 PathWithMetrics {
48 path,
49 cost: metrics.cost,
50 total_latency_ms: metrics.total_latency_ms,
51 min_probe_success_rate: metrics.min_probe_success_rate,
52 min_ack_rate: metrics.min_ack_rate,
53 fundable_tickets_floor: metrics.fundable_tickets_floor,
54 }
55 }
56}
57
58fn opt_min<T: PartialOrd>(a: Option<T>, b: Option<T>) -> Option<T> {
60 match (a, b) {
61 (Some(x), Some(y)) => Some(if x <= y { x } else { y }),
62 (x, y) => x.or(y),
63 }
64}
65
66struct MetricsValueFn<W: EdgeObservableRead> {
73 inner: EdgeValueFn<f64, W>,
74 ticket_face_value: Option<hopr_api::graph::traits::Balance>,
80}
81
82impl<W> ValueFn for MetricsValueFn<W>
83where
84 W: EdgeObservableRead + Send + 'static,
85{
86 type Value = PathCostWithMetrics;
87 type Weight = W;
88
89 fn initial_value(&self) -> Self::Value {
90 PathCostWithMetrics {
91 cost: self.inner.initial_value(),
92 total_latency_ms: Some(0),
93 min_probe_success_rate: None,
94 min_ack_rate: None,
95 fundable_tickets_floor: None,
96 }
97 }
98
99 fn min_value(&self) -> Option<Self::Value> {
100 self.inner.min_value().map(|c| PathCostWithMetrics {
101 cost: c,
102 total_latency_ms: None,
103 min_probe_success_rate: None,
104 min_ack_rate: None,
105 fundable_tickets_floor: None,
106 })
107 }
108
109 fn into_value_fn(self) -> BasicValueFn<Self::Value, Self::Weight> {
110 let inner = self.inner.into_value_fn();
111 let ticket_face_value = self.ticket_face_value;
112 Arc::new(move |prev: PathCostWithMetrics, observed: &W, idx: usize| {
113 let cost = inner(prev.cost, observed, idx);
114
115 let edge_lat = observed
116 .immediate_qos()
117 .and_then(|m| m.average_latency())
118 .or_else(|| observed.intermediate_qos().and_then(|m| m.average_latency()));
119 let total_latency_ms = match (prev.total_latency_ms, edge_lat) {
120 (Some(acc), Some(lat)) => Some(((acc as u128 + lat.as_millis()).min(u32::MAX as u128)) as u32),
121 _ => None,
122 };
123
124 let edge_probe = observed
129 .immediate_qos()
130 .and_then(|m| m.average_probe_rate())
131 .into_iter()
132 .chain(observed.intermediate_qos().and_then(|m| m.average_probe_rate()))
133 .reduce(f64::min);
134 let min_probe_success_rate = opt_min(prev.min_probe_success_rate, edge_probe);
135
136 let edge_ack = observed.immediate_qos().and_then(|m| m.ack_rate());
137 let min_ack_rate = opt_min(prev.min_ack_rate, edge_ack);
138
139 let edge_tickets = observed
148 .intermediate_qos()
149 .and_then(|m| m.balance())
150 .zip(ticket_face_value)
151 .map(|(balance, face_value)| {
152 let tickets = if face_value.is_zero() {
153 hopr_api::graph::traits::Balance::zero()
154 } else {
155 balance / face_value
156 };
157 if tickets > hopr_api::graph::traits::Balance::from(u128::MAX) {
158 u128::MAX
159 } else {
160 tickets.low_u128()
161 }
162 });
163 let fundable_tickets_floor = opt_min(prev.fundable_tickets_floor, edge_tickets);
164
165 PathCostWithMetrics {
166 cost,
167 total_latency_ms,
168 min_probe_success_rate,
169 min_ack_rate,
170 fundable_tickets_floor,
171 }
172 })
173 }
174}
175
176pub fn prune_for_consistency(candidates: Vec<PathWithMetrics>, floor: usize, hops: usize) -> Vec<PathWithMetrics> {
191 if floor == 0 || candidates.len() <= floor {
193 return candidates;
194 }
195
196 let fully_measured =
197 |p: &PathWithMetrics| p.total_latency_ms.is_some() && (hops == 0 || p.fundable_tickets_floor.is_some());
198
199 let (mut populated, unpopulated): (Vec<_>, Vec<_>) = candidates.into_iter().partition(|p| fully_measured(p));
200
201 populated.sort_by_key(|p| p.total_latency_ms.unwrap_or(u32::MAX));
203
204 let target_populated = populated.len().min(floor);
209 let populated = take_relayer_diverse(populated, target_populated);
210 let remaining = floor - target_populated;
211
212 let mut result = populated;
213 result.extend(unpopulated.into_iter().take(remaining));
214
215 result
216}
217
218fn take_relayer_diverse(sorted: Vec<PathWithMetrics>, n: usize) -> Vec<PathWithMetrics> {
224 if sorted.len() <= n {
225 return sorted;
226 }
227
228 let mut seen: Vec<OffchainPublicKey> = Vec::with_capacity(n);
229 let mut taken = vec![false; sorted.len()];
230 let mut count = 0;
231
232 for (i, p) in sorted.iter().enumerate() {
234 if count == n {
235 break;
236 }
237 let Some(first) = p.path.first() else { continue };
238 if !seen.contains(first) {
239 seen.push(*first);
240 taken[i] = true;
241 count += 1;
242 }
243 }
244
245 for slot in taken.iter_mut() {
247 if count == n {
248 break;
249 }
250 if !*slot {
251 *slot = true;
252 count += 1;
253 }
254 }
255
256 sorted
257 .into_iter()
258 .zip(taken)
259 .filter_map(|(p, keep)| keep.then_some(p))
260 .collect()
261}
262
263pub(crate) fn distinct_first_relayers(paths: &[PathWithMetrics]) -> usize {
270 paths
271 .iter()
272 .filter_map(|p| p.path.first())
273 .collect::<std::collections::HashSet<_>>()
274 .len()
275}
276
277fn compute_paths<G, W>(
284 graph: &G,
285 src: &OffchainPublicKey,
286 dest: &OffchainPublicKey,
287 length: std::num::NonZeroUsize,
288 take: usize,
289 value_fn: MetricsValueFn<W>,
290) -> Vec<PathWithMetrics>
291where
292 G: NetworkGraphTraverse<NodeId = OffchainPublicKey, Observed = W>,
293 W: EdgeObservableRead + Send + 'static,
294{
295 let raw = graph.simple_paths(src, dest, length.get(), Some(take), value_fn);
296
297 raw.into_iter()
298 .filter_map(|(path, _, metrics)| {
299 tracing::trace!(?path, cost = metrics.cost, "evaluating candidate path");
300 if metrics.cost > 0.0 {
301 let mut path = path;
302 path.push(*dest);
303 Some(PathWithMetrics::from((metrics, path)))
304 } else {
305 None
306 }
307 })
308 .collect()
309}
310
311#[derive(Clone)]
323pub struct HoprGraphPathSelector<G> {
324 me: OffchainPublicKey,
325 graph: G,
326 max_paths: usize,
327 edge_penalty: f64,
328 min_ack_rate: f64,
329 anonymity_floor: usize,
330}
331
332impl<G> HoprGraphPathSelector<G>
333where
334 G: NetworkGraphTraverse<NodeId = OffchainPublicKey>
335 + NetworkGraphView<NodeId = OffchainPublicKey>
336 + Clone
337 + Send
338 + Sync
339 + 'static,
340 <G as NetworkGraphTraverse>::Observed: EdgeObservableRead + Send + 'static,
341{
342 pub fn new(
351 me: OffchainPublicKey,
352 graph: G,
353 max_paths: usize,
354 edge_penalty: f64,
355 min_ack_rate: f64,
356 anonymity_floor: usize,
357 ) -> Self {
358 Self {
359 me,
360 graph,
361 max_paths,
362 edge_penalty,
363 min_ack_rate,
364 anonymity_floor,
365 }
366 }
367
368 fn compute_extended_forward_paths(
378 &self,
379 src: &OffchainPublicKey,
380 dest: &OffchainPublicKey,
381 shorter_length: std::num::NonZeroUsize,
382 take: usize,
383 existing: &[PathWithMetrics],
384 ticket_face_value: Option<hopr_api::graph::traits::Balance>,
386 ) -> Vec<PathWithMetrics> {
387 let value_fn = MetricsValueFn {
388 inner: EdgeValueFn::forward_without_self_loopback(
389 shorter_length,
390 self.edge_penalty,
391 self.min_ack_rate,
392 ticket_face_value,
393 ),
394 ticket_face_value,
395 };
396 let raw = self
397 .graph
398 .simple_paths_from(src, shorter_length.get(), Some(take), value_fn);
399
400 raw.into_iter()
401 .filter_map(|(path, _, metrics)| {
402 if metrics.cost <= 0.0 {
403 return None;
404 }
405
406 if path.contains(dest) {
410 return None;
411 }
412 let mut candidate = path;
413 candidate.push(*dest);
414
415 if existing.iter().any(|pwm| pwm.path == candidate) {
416 return None;
417 }
418
419 tracing::trace!(?candidate, cost = metrics.cost, "extended forward path candidate");
420 Some(PathWithMetrics::from((metrics, candidate)))
421 })
422 .take(take)
423 .collect()
424 }
425}
426
427impl<G> PathSelector for HoprGraphPathSelector<G>
428where
429 G: NetworkGraphTraverse<NodeId = OffchainPublicKey>
430 + NetworkGraphView<NodeId = OffchainPublicKey>
431 + Clone
432 + Send
433 + Sync
434 + 'static,
435 <G as NetworkGraphTraverse>::Observed: EdgeObservableRead + Send + 'static,
436{
437 #[tracing::instrument(level = "trace", skip(self), fields(src = %src, dest = %dest, hops), ret, err)]
449 fn select_path(
450 &self,
451 src: OffchainPublicKey,
452 dest: OffchainPublicKey,
453 hops: usize,
454 ) -> Result<Vec<PathWithMetrics>> {
455 let direction = if src == self.me { "forward" } else { "return" };
456 tracing::debug!(%src, %dest, hops, direction, "computing paths from graph");
457
458 let length = std::num::NonZeroUsize::new(hops + 1)
459 .expect("can never fail, it is physically at least 1 after the addition");
460
461 let ticket_face_value = self.graph.ticket_face_value();
465
466 let paths = if src == self.me {
467 let mut found = compute_paths(
469 &self.graph,
470 &src,
471 &dest,
472 length,
473 self.max_paths,
474 MetricsValueFn {
475 inner: EdgeValueFn::forward(length, self.edge_penalty, self.min_ack_rate, ticket_face_value),
476 ticket_face_value,
477 },
478 );
479 tracing::debug!(
480 direction,
481 phase = 1,
482 count = found.len(),
483 "[forward] phase 1 candidates"
484 );
485
486 if found.len() < self.max_paths
489 && let Some(shorter) = std::num::NonZeroUsize::new(length.get() - 1)
490 {
491 let remaining = self.max_paths - found.len();
492 let extended =
493 self.compute_extended_forward_paths(&src, &dest, shorter, remaining, &found, ticket_face_value);
494 tracing::debug!(
495 direction,
496 phase = 2,
497 count = extended.len(),
498 "[forward] phase 2 extended candidates"
499 );
500 found.extend(extended);
501 }
502
503 found
504 } else {
505 let found = compute_paths(
506 &self.graph,
507 &src,
508 &dest,
509 length,
510 self.max_paths,
511 MetricsValueFn {
512 inner: EdgeValueFn::returning(length, self.edge_penalty, self.min_ack_rate, ticket_face_value),
513 ticket_face_value,
514 },
515 );
516 tracing::debug!(direction, count = found.len(), "[return] candidates");
517 found
518 };
519
520 for (i, pwm) in paths.iter().enumerate() {
521 tracing::debug!(
522 direction,
523 index = i,
524 path = ?pwm.path,
525 cost = pwm.cost,
526 total_latency_ms = ?pwm.total_latency_ms,
527 "[{direction}] candidate path"
528 );
529 }
530
531 if paths.is_empty() {
532 return Err(PathPlannerError::Path(PathError::PathNotFound(
533 hops,
534 src.to_string(),
535 dest.to_string(),
536 )));
537 }
538
539 let pruned = prune_for_consistency(paths, self.anonymity_floor, hops);
540 let relayer_set = pruned
541 .iter()
542 .filter_map(|p| p.path.first())
543 .collect::<std::collections::HashSet<_>>();
544 tracing::debug!(
545 %src,
546 %dest,
547 hops,
548 direction,
549 survived = pruned.len(),
550 distinct_relayers = relayer_set.len(),
551 first_relayers = ?relayer_set,
552 "pruned candidate paths",
553 );
554
555 Ok(pruned)
556 }
557}
558
559#[cfg(test)]
560mod tests {
561 use std::time::Duration;
562
563 use anyhow::Context;
564 use hex_literal::hex;
565 use hopr_api::{
566 graph::{
567 NetworkGraphUpdate, NetworkGraphWrite,
568 traits::{EdgeObservableWrite, EdgeWeightType},
569 },
570 types::{
571 crypto::prelude::{Keypair, OffchainKeypair},
572 internal::routing::RoutingOptions,
573 },
574 };
575 use hopr_network_graph::ChannelGraph;
576
577 use super::*;
578 use crate::path::{PathPlannerConfig, traits::PathSelector};
579
580 fn test_selector(
581 me: OffchainPublicKey,
582 graph: ChannelGraph,
583 max_paths: usize,
584 ) -> HoprGraphPathSelector<ChannelGraph> {
585 let cfg = PathPlannerConfig::default();
586 HoprGraphPathSelector::new(
587 me,
588 graph,
589 max_paths,
590 cfg.edge_penalty,
591 cfg.min_ack_rate,
592 cfg.min_paths_anonymity_floor,
593 )
594 }
595
596 const SECRET_0: [u8; 32] = hex!("60741b83b99e36aa0c1331578156e16b8e21166d01834abb6c64b103f885734d");
597 const SECRET_1: [u8; 32] = hex!("71bf1f42ebbfcd89c3e197a3fd7cda79b92499e509b6fefa0fe44d02821d146a");
598 const SECRET_2: [u8; 32] = hex!("c24bd833704dd2abdae3933fcc9962c2ac404f84132224c474147382d4db2299");
599 const SECRET_3: [u8; 32] = hex!("e0bf93e9c916104da00b1850adc4608bd7e9087bbd3f805451f4556aa6b3fd6e");
600 const SECRET_4: [u8; 32] = hex!("cfc66f718ec66fb822391775d749d7a0d66b690927673634816b63339bc12a3c");
601
602 const MAX_PATHS: usize = 4;
603
604 fn pubkey(secret: &[u8; 32]) -> OffchainPublicKey {
605 *OffchainKeypair::from_secret(secret).expect("valid secret").public()
606 }
607
608 fn mark_edge_full(graph: &ChannelGraph, src: &OffchainPublicKey, dst: &OffchainPublicKey) {
610 graph.upsert_edge(src, dst, |obs| {
611 obs.record(EdgeWeightType::Connected(true));
612 obs.record(EdgeWeightType::Immediate(Ok(Duration::from_millis(50))));
613 obs.record(EdgeWeightType::Intermediate(Ok(Duration::from_millis(50))));
614 obs.record(EdgeWeightType::Balance(Some(hopr_api::graph::traits::Balance::from(
615 1000u64,
616 ))));
617 });
618 }
619
620 fn two_hop_graph() -> (OffchainPublicKey, OffchainPublicKey, OffchainPublicKey, ChannelGraph) {
622 let me = pubkey(&SECRET_0);
623 let hop = pubkey(&SECRET_1);
624 let dest = pubkey(&SECRET_2);
625 let graph = ChannelGraph::new(me);
626 graph.add_node(hop);
627 graph.add_node(dest);
628 graph.add_edge(&me, &hop).unwrap();
630 graph.add_edge(&hop, &dest).unwrap();
631 mark_edge_full(&graph, &me, &hop);
632 mark_edge_full(&graph, &hop, &dest);
633 graph.add_edge(&dest, &hop).unwrap();
635 graph.add_edge(&hop, &me).unwrap();
636 mark_edge_full(&graph, &dest, &hop);
637 mark_edge_full(&graph, &hop, &me);
638 (me, hop, dest, graph)
639 }
640
641 #[tokio::test]
642 async fn unreachable_dest_should_return_error() -> anyhow::Result<()> {
643 let me = pubkey(&SECRET_0);
644 let unreachable = pubkey(&SECRET_1);
645 let graph = ChannelGraph::new(me);
646 let selector = test_selector(me, graph, MAX_PATHS);
648
649 let fwd = selector.select_path(me, unreachable, 1);
650 assert!(fwd.is_err(), "forward: should error when destination is unreachable");
651 assert!(matches!(
652 fwd.unwrap_err(),
653 PathPlannerError::Path(PathError::PathNotFound(..))
654 ));
655
656 let rev = selector.select_path(unreachable, me, 1);
657 assert!(rev.is_err(), "reverse: should error when destination is unreachable");
658 assert!(matches!(
659 rev.unwrap_err(),
660 PathPlannerError::Path(PathError::PathNotFound(..))
661 ));
662
663 Ok(())
664 }
665
666 #[tokio::test]
667 async fn path_should_exclude_source() -> anyhow::Result<()> {
668 let (me, _hop, dest, graph) = two_hop_graph();
669 let selector = test_selector(me, graph, MAX_PATHS);
670
671 let fwd = selector.select_path(me, dest, 1).context("forward path")?;
672 assert!(!fwd.is_empty());
673 for pwm in &fwd {
674 assert!(!pwm.path.contains(&me), "forward path must not contain the source");
675 assert!(pwm.cost > 0.0, "cost must be positive");
676 }
677
678 let rev = selector.select_path(dest, me, 1).context("reverse path")?;
679 assert!(!rev.is_empty());
680 for pwm in &rev {
681 assert!(!pwm.path.contains(&dest), "reverse path must not contain the source");
682 assert!(pwm.cost > 0.0, "cost must be positive");
683 }
684
685 Ok(())
686 }
687
688 #[tokio::test]
689 async fn multi_hop_path_should_have_correct_length() -> anyhow::Result<()> {
690 let me = pubkey(&SECRET_0);
692 let a = pubkey(&SECRET_1);
693 let b = pubkey(&SECRET_2);
694 let dest = pubkey(&SECRET_3);
695 let graph = ChannelGraph::new(me);
696 for n in [a, b, dest] {
697 graph.add_node(n);
698 }
699 graph.add_edge(&me, &a).unwrap();
701 graph.add_edge(&a, &b).unwrap();
702 graph.add_edge(&b, &dest).unwrap();
703 mark_edge_full(&graph, &me, &a);
704 mark_edge_full(&graph, &a, &b);
705 mark_edge_full(&graph, &b, &dest);
706 graph.add_edge(&dest, &b).unwrap();
708 graph.add_edge(&b, &a).unwrap();
709 graph.add_edge(&a, &me).unwrap();
710 mark_edge_full(&graph, &dest, &b);
711 mark_edge_full(&graph, &b, &a);
712 mark_edge_full(&graph, &a, &me);
713
714 let selector = test_selector(me, graph, MAX_PATHS);
715
716 let fwd = selector.select_path(me, dest, 2).context("forward 2-hop path")?;
717 assert!(!fwd.is_empty());
718 for pwm in &fwd {
719 assert_eq!(pwm.path.len(), 3, "forward 2-hop path: [A, B, dest]");
720 assert_eq!(pwm.path.last(), Some(&dest));
721 }
722
723 let rev = selector.select_path(dest, me, 2).context("reverse 2-hop path")?;
724 assert!(!rev.is_empty());
725 for pwm in &rev {
726 assert_eq!(pwm.path.len(), 3, "reverse 2-hop path: [B, A, me]");
727 assert_eq!(pwm.path.last(), Some(&me));
728 }
729
730 Ok(())
731 }
732
733 #[tokio::test]
734 async fn one_hop_path_should_include_relay_and_destination() -> anyhow::Result<()> {
735 let (me, relay, dest, graph) = two_hop_graph();
737 let selector = test_selector(me, graph, MAX_PATHS);
738
739 let fwd = selector.select_path(me, dest, 1).context("forward 1-hop path")?;
740 assert!(!fwd.is_empty());
741 for pwm in &fwd {
742 assert_eq!(pwm.path.len(), 2, "forward: [relay, dest]");
743 assert_eq!(pwm.path.last(), Some(&dest));
744 assert!(!pwm.path.contains(&me));
745 }
746
747 let rev = selector.select_path(dest, me, 1).context("reverse 1-hop path")?;
748 assert!(!rev.is_empty());
749 for pwm in &rev {
750 assert_eq!(pwm.path.len(), 2, "reverse: [relay, me]");
751 assert_eq!(pwm.path.last(), Some(&me));
752 assert!(!pwm.path.contains(&dest));
753 }
754
755 let _ = relay;
756 Ok(())
757 }
758
759 #[tokio::test]
760 async fn diamond_topology_should_return_multiple_paths() -> anyhow::Result<()> {
761 let me = pubkey(&SECRET_0);
763 let a = pubkey(&SECRET_1);
764 let b = pubkey(&SECRET_2);
765 let dest = pubkey(&SECRET_3);
766 let graph = ChannelGraph::new(me);
767 for n in [a, b, dest] {
768 graph.add_node(n);
769 }
770 graph.add_edge(&me, &a).unwrap();
772 graph.add_edge(&me, &b).unwrap();
773 graph.add_edge(&a, &dest).unwrap();
774 graph.add_edge(&b, &dest).unwrap();
775 mark_edge_full(&graph, &me, &a);
776 mark_edge_full(&graph, &me, &b);
777 mark_edge_full(&graph, &a, &dest);
778 mark_edge_full(&graph, &b, &dest);
779 graph.add_edge(&dest, &a).unwrap();
781 graph.add_edge(&dest, &b).unwrap();
782 graph.add_edge(&a, &me).unwrap();
783 graph.add_edge(&b, &me).unwrap();
784 mark_edge_full(&graph, &dest, &a);
785 mark_edge_full(&graph, &dest, &b);
786 mark_edge_full(&graph, &a, &me);
787 mark_edge_full(&graph, &b, &me);
788
789 let selector = test_selector(me, graph, MAX_PATHS);
790
791 let fwd = selector.select_path(me, dest, 1).context("forward path")?;
792 assert_eq!(fwd.len(), 2, "forward: both paths via a and b should be returned");
793 for pwm in &fwd {
794 assert_eq!(pwm.path.last(), Some(&dest));
795 }
796
797 let rev = selector.select_path(dest, me, 1).context("reverse path")?;
798 assert_eq!(rev.len(), 2, "reverse: both paths via a and b should be returned");
799 for pwm in &rev {
800 assert_eq!(pwm.path.last(), Some(&me));
801 }
802
803 Ok(())
804 }
805
806 #[tokio::test]
807 async fn zero_cost_paths_should_return_error() -> anyhow::Result<()> {
808 let me = pubkey(&SECRET_0);
810 let dest = pubkey(&SECRET_1);
811 let graph = ChannelGraph::new(me);
812 graph.add_node(dest);
813 graph.add_edge(&me, &dest).unwrap();
814 graph.add_edge(&dest, &me).unwrap();
815 let selector = test_selector(me, graph, MAX_PATHS);
818 assert!(
819 selector.select_path(me, dest, 1).is_err(),
820 "forward: edge with no observations should produce no valid path"
821 );
822 assert!(
823 selector.select_path(dest, me, 1).is_err(),
824 "reverse: edge with no observations should produce no valid path"
825 );
826 Ok(())
827 }
828
829 #[tokio::test]
830 async fn no_path_at_requested_hop_count_should_return_error() -> anyhow::Result<()> {
831 let me = pubkey(&SECRET_0);
833 let dest = pubkey(&SECRET_1);
834 let graph = ChannelGraph::new(me);
835 graph.add_node(dest);
836 graph.add_edge(&me, &dest).unwrap();
837 graph.add_edge(&dest, &me).unwrap();
838 mark_edge_full(&graph, &me, &dest);
839 mark_edge_full(&graph, &dest, &me);
840
841 let selector = test_selector(me, graph, MAX_PATHS);
842 assert!(
843 selector.select_path(me, dest, 2).is_err(),
844 "forward: no 2-hop path should exist for a direct edge"
845 );
846 assert!(
847 selector.select_path(dest, me, 2).is_err(),
848 "reverse: no 2-hop path should exist for a direct edge"
849 );
850 Ok(())
851 }
852
853 #[tokio::test]
854 async fn forward_path_should_work_without_last_edge() -> anyhow::Result<()> {
855 let me = pubkey(&SECRET_0);
859 let relay = pubkey(&SECRET_1);
860 let dest = pubkey(&SECRET_2);
861 let graph = ChannelGraph::new(me);
862 graph.add_node(relay);
863 graph.add_node(dest);
864 graph.add_edge(&me, &relay).unwrap();
866 mark_edge_full(&graph, &me, &relay);
867 graph.add_edge(&dest, &relay).unwrap();
869 graph.add_edge(&relay, &me).unwrap();
870 mark_edge_full(&graph, &dest, &relay);
871 mark_edge_full(&graph, &relay, &me);
872
873 let selector = test_selector(me, graph, MAX_PATHS);
874
875 let fwd = selector
877 .select_path(me, dest, 1)
878 .context("forward path with virtual last hop")?;
879 assert!(!fwd.is_empty(), "forward path should find at least one route");
880 for pwm in &fwd {
881 assert_eq!(pwm.path.len(), 2, "forward: [relay, dest]");
882 assert_eq!(pwm.path[0], relay);
883 assert_eq!(pwm.path[1], dest);
884 }
885
886 let rev = selector.select_path(dest, me, 1).context("return path")?;
888 assert!(!rev.is_empty(), "return path should find at least one route");
889 for pwm in &rev {
890 assert_eq!(pwm.path.len(), 2, "return: [relay, me]");
891 assert_eq!(pwm.path.last(), Some(&me));
892 }
893
894 Ok(())
895 }
896
897 #[tokio::test]
898 async fn five_node_chain_should_support_max_hops() -> anyhow::Result<()> {
899 let me = pubkey(&SECRET_0);
901 let a = pubkey(&SECRET_1);
902 let b = pubkey(&SECRET_2);
903 let c = pubkey(&SECRET_3);
904 let dest = pubkey(&SECRET_4);
905 let graph = ChannelGraph::new(me);
906 for n in [a, b, c, dest] {
907 graph.add_node(n);
908 }
909 graph.add_edge(&me, &a).unwrap();
911 graph.add_edge(&a, &b).unwrap();
912 graph.add_edge(&b, &c).unwrap();
913 graph.add_edge(&c, &dest).unwrap();
914 mark_edge_full(&graph, &me, &a);
915 mark_edge_full(&graph, &a, &b);
916 mark_edge_full(&graph, &b, &c);
917 mark_edge_full(&graph, &c, &dest);
918 graph.add_edge(&dest, &c).unwrap();
920 graph.add_edge(&c, &b).unwrap();
921 graph.add_edge(&b, &a).unwrap();
922 graph.add_edge(&a, &me).unwrap();
923 mark_edge_full(&graph, &dest, &c);
924 mark_edge_full(&graph, &c, &b);
925 mark_edge_full(&graph, &b, &a);
926 mark_edge_full(&graph, &a, &me);
927
928 let selector = test_selector(me, graph, MAX_PATHS);
929
930 let fwd = selector
931 .select_path(me, dest, RoutingOptions::MAX_INTERMEDIATE_HOPS)
932 .context("forward 3-hop path")?;
933 assert!(!fwd.is_empty());
934 for pwm in &fwd {
935 assert_eq!(pwm.path.len(), 4, "forward: [a, b, c, dest]");
936 assert_eq!(pwm.path.last(), Some(&dest));
937 assert!(!pwm.path.contains(&me));
938 }
939
940 let rev = selector
941 .select_path(dest, me, RoutingOptions::MAX_INTERMEDIATE_HOPS)
942 .context("reverse 3-hop path")?;
943 assert!(!rev.is_empty());
944 for pwm in &rev {
945 assert_eq!(pwm.path.len(), 4, "reverse: [c, b, a, me]");
946 assert_eq!(pwm.path.last(), Some(&me));
947 assert!(!pwm.path.contains(&dest));
948 }
949
950 Ok(())
951 }
952
953 #[tokio::test]
954 async fn selector_should_reject_extended_path_containing_destination() -> anyhow::Result<()> {
955 let me = pubkey(&SECRET_0);
960 let relay = pubkey(&SECRET_1);
961 let dest = pubkey(&SECRET_2);
962 let graph = ChannelGraph::new(me);
963 graph.add_node(relay);
964 graph.add_node(dest);
965 graph.add_edge(&me, &dest).unwrap();
967 mark_edge_full(&graph, &me, &dest);
968 graph.add_edge(&me, &relay).unwrap();
970 mark_edge_full(&graph, &me, &relay);
971 graph.add_edge(&dest, &relay).unwrap();
973 graph.add_edge(&relay, &me).unwrap();
974 mark_edge_full(&graph, &dest, &relay);
975 mark_edge_full(&graph, &relay, &me);
976
977 let selector = test_selector(me, graph, MAX_PATHS);
978
979 let fwd = selector
980 .select_path(me, dest, 1)
981 .context("forward path with dest as direct neighbor")?;
982 assert!(!fwd.is_empty(), "should find at least one path via relay");
983 for pwm in &fwd {
984 assert_eq!(pwm.path.len(), 2, "path must be [relay, dest]");
985 assert_eq!(pwm.path[0], relay, "first node must be relay, not dest");
986 assert_eq!(pwm.path[1], dest);
987 }
988 Ok(())
989 }
990
991 #[tokio::test]
992 async fn selector_should_reject_one_hop_path_where_relay_equals_destination() -> anyhow::Result<()> {
993 let me = pubkey(&SECRET_0);
996 let relay = pubkey(&SECRET_1);
997 let dest = pubkey(&SECRET_2);
998 let graph = ChannelGraph::new(me);
999 graph.add_node(relay);
1000 graph.add_node(dest);
1001 graph.add_edge(&me, &dest).unwrap();
1003 mark_edge_full(&graph, &me, &dest);
1004 graph.add_edge(&me, &relay).unwrap();
1006 mark_edge_full(&graph, &me, &relay);
1007 graph.add_edge(&dest, &relay).unwrap();
1009 graph.add_edge(&relay, &me).unwrap();
1010 mark_edge_full(&graph, &dest, &relay);
1011 mark_edge_full(&graph, &relay, &me);
1012
1013 let selector = test_selector(me, graph, MAX_PATHS);
1014
1015 let fwd = selector
1016 .select_path(me, dest, 1)
1017 .context("forward path — dest is direct neighbor, relay is intermediate")?;
1018 assert!(!fwd.is_empty(), "should find path via relay (virtual last hop)");
1019 for pwm in &fwd {
1020 assert_eq!(pwm.path[0], relay, "intermediate must be relay, not dest");
1021 assert_ne!(pwm.path[0], dest, "dest must not appear as intermediate");
1022 }
1023 Ok(())
1024 }
1025
1026 #[tokio::test]
1027 async fn selector_should_skip_zero_cost_paths() -> anyhow::Result<()> {
1028 let me = pubkey(&SECRET_0);
1030 let hop = pubkey(&SECRET_1);
1031 let dest = pubkey(&SECRET_2);
1032 let graph = ChannelGraph::new(me);
1033 graph.add_node(hop);
1034 graph.add_node(dest);
1035 graph.add_edge(&me, &hop).context("adding edge me -> hop")?;
1036 graph.add_edge(&hop, &dest).context("adding edge hop -> dest")?;
1037 let selector = test_selector(me, graph, MAX_PATHS);
1040
1041 let err = selector
1042 .select_path(me, dest, 1)
1043 .expect_err("zero-cost paths should be filtered out");
1044 anyhow::ensure!(
1045 matches!(err, PathPlannerError::Path(PathError::PathNotFound(..))),
1046 "expected PathNotFound, got: {err}"
1047 );
1048 Ok(())
1049 }
1050
1051 fn make_path_with_latency(latency_ms: Option<u32>) -> PathWithMetrics {
1054 PathWithMetrics {
1055 path: vec![],
1056 cost: 1.0,
1057 total_latency_ms: latency_ms,
1058 min_probe_success_rate: None,
1059 min_ack_rate: None,
1060 fundable_tickets_floor: None,
1061 }
1062 }
1063
1064 fn make_path_with_tickets(latency_ms: Option<u32>, fundable_tickets_floor: Option<u128>) -> PathWithMetrics {
1065 PathWithMetrics {
1066 path: vec![],
1067 cost: 1.0,
1068 total_latency_ms: latency_ms,
1069 min_probe_success_rate: None,
1070 min_ack_rate: None,
1071 fundable_tickets_floor,
1072 }
1073 }
1074
1075 fn make_path_via(relayer_idx: u8, latency_ms: u32) -> PathWithMetrics {
1077 let mut secret = [1u8; 32];
1078 secret[0] = relayer_idx.max(1);
1079 let relayer = *OffchainKeypair::from_secret(&secret).expect("valid secret").public();
1080 PathWithMetrics {
1081 path: vec![relayer],
1082 cost: 1.0,
1083 total_latency_ms: Some(latency_ms),
1084 min_probe_success_rate: None,
1085 min_ack_rate: None,
1086 fundable_tickets_floor: Some(1000),
1087 }
1088 }
1089
1090 #[test]
1091 fn prune_should_prefer_distinct_first_relayers_over_pure_latency_order() {
1092 let candidates = vec![
1095 make_path_via(1, 10),
1096 make_path_via(1, 11),
1097 make_path_via(1, 12),
1098 make_path_via(1, 13),
1099 make_path_via(2, 50),
1100 make_path_via(3, 60),
1101 make_path_via(4, 70),
1102 ];
1103
1104 let result = prune_for_consistency(candidates, 3, 1);
1105 assert_eq!(3, result.len());
1106
1107 let relayers: Vec<String> = result
1108 .iter()
1109 .map(|p| hopr_api::types::primitive::traits::ToHex::to_hex(p.path.first().expect("non-empty")))
1110 .collect();
1111 let mut distinct = relayers.clone();
1112 distinct.sort();
1113 distinct.dedup();
1114 assert_eq!(3, distinct.len(), "the floor must be filled with distinct relayers");
1115
1116 assert_eq!(Some(10), result[0].total_latency_ms, "still latency-ordered");
1119 }
1120
1121 #[test]
1122 fn distinct_first_relayers_counts_unique_first_hops() {
1123 let paths = vec![
1126 make_path_via(1, 10),
1127 make_path_via(1, 20),
1128 make_path_via(2, 30),
1129 make_path_with_latency(Some(40)), ];
1131 assert_eq!(2, distinct_first_relayers(&paths));
1132 }
1133
1134 #[test]
1135 fn floor_zero_keeps_all_distinct_relayers_unlike_a_small_cap() {
1136 let candidates: Vec<_> = (1..=10u8).map(|i| make_path_via(i, i as u32 * 10)).collect();
1141
1142 let uncapped = prune_for_consistency(candidates.clone(), 0, 1);
1143 assert_eq!(10, uncapped.len(), "floor = 0 disables pruning: all candidates survive");
1144 assert_eq!(10, distinct_first_relayers(&uncapped), "all ten relayers kept");
1145
1146 let capped = prune_for_consistency(candidates, 2, 1);
1147 assert_eq!(2, capped.len(), "floor = 2 caps survivors at two");
1148 assert_eq!(2, distinct_first_relayers(&capped));
1149 }
1150
1151 #[test]
1152 fn prune_should_fall_back_to_latency_order_once_relayers_are_exhausted() {
1153 let candidates = vec![
1156 make_path_via(1, 10),
1157 make_path_via(2, 20),
1158 make_path_via(1, 30),
1159 make_path_via(2, 40),
1160 make_path_via(1, 50),
1161 ];
1162
1163 let result = prune_for_consistency(candidates, 4, 1);
1164 assert_eq!(4, result.len());
1165 assert_eq!(
1166 vec![Some(10), Some(20), Some(30), Some(40)],
1167 result.iter().map(|p| p.total_latency_ms).collect::<Vec<_>>()
1168 );
1169 }
1170
1171 #[test]
1172 fn prune_keeps_all_when_below_floor() {
1173 let candidates: Vec<_> = (0..5).map(|i| make_path_with_latency(Some(i * 10))).collect();
1174 let result = prune_for_consistency(candidates, 8, 1);
1175 assert_eq!(result.len(), 5, "below floor: nothing should be dropped");
1176 }
1177
1178 #[test]
1179 fn prune_drops_high_latency_first() {
1180 let candidates: Vec<_> = (0..30u32)
1182 .map(|i| make_path_with_tickets(Some(i * 10), Some(1_000_000)))
1183 .collect();
1184 let result = prune_for_consistency(candidates, 8, 1);
1185 assert_eq!(result.len(), 8);
1186 for p in &result {
1187 assert!(p.total_latency_ms.unwrap() < 80, "only the 8 lowest should survive");
1188 }
1189 }
1190
1191 #[test]
1192 fn prune_preserves_populated_paths_over_unpopulated() {
1193 let mut candidates: Vec<_> = vec![
1197 make_path_with_tickets(Some(10), Some(1_000)),
1198 make_path_with_tickets(Some(30), Some(1_000)),
1199 make_path_with_tickets(Some(20), Some(1_000)),
1200 ];
1201 candidates.extend((0..6).map(|_| make_path_with_latency(None)));
1202 let result = prune_for_consistency(candidates, 8, 1);
1203 assert_eq!(result.len(), 8);
1204 let populated: Vec<_> = result.iter().filter(|p| p.total_latency_ms.is_some()).collect();
1206 assert_eq!(populated.len(), 3);
1207 assert!(populated.iter().any(|p| p.total_latency_ms == Some(10)));
1208 assert!(populated.iter().any(|p| p.total_latency_ms == Some(20)));
1209 assert!(populated.iter().any(|p| p.total_latency_ms == Some(30)));
1210 }
1211
1212 #[test]
1213 fn prune_drops_unpopulated_when_all_populated_exhausted() {
1214 let candidates: Vec<_> = (0..20).map(|_| make_path_with_latency(None)).collect();
1216 let result = prune_for_consistency(candidates, 8, 1);
1217 assert_eq!(result.len(), 8);
1218 }
1219
1220 #[test]
1221 fn prune_keeps_populated_when_unpopulated_exceeds_floor() {
1222 let mut candidates: Vec<_> = vec![
1226 make_path_with_tickets(Some(10), Some(1_000)),
1227 make_path_with_tickets(Some(20), Some(1_000)),
1228 ];
1229 candidates.extend((0..10).map(|_| make_path_with_latency(None)));
1230 let result = prune_for_consistency(candidates, 8, 1);
1231 assert_eq!(result.len(), 8);
1232 let populated: Vec<_> = result.iter().filter(|p| p.total_latency_ms.is_some()).collect();
1233 assert_eq!(populated.len(), 2, "both measured paths must survive");
1234 assert!(populated.iter().any(|p| p.total_latency_ms == Some(10)));
1235 assert!(populated.iter().any(|p| p.total_latency_ms == Some(20)));
1236 }
1237
1238 #[test]
1239 fn prune_exact_floor_is_unchanged() {
1240 let candidates: Vec<_> = (0..8)
1241 .map(|i| make_path_with_tickets(Some(i * 10), Some(1_000)))
1242 .collect();
1243 let result = prune_for_consistency(candidates, 8, 1);
1244 assert_eq!(result.len(), 8);
1245 }
1246
1247 #[test]
1248 fn prune_0_hop_with_measured_latency_and_no_capacity_is_populated() {
1249 let mut candidates: Vec<_> = vec![
1252 make_path_with_tickets(Some(50), None), ];
1254 candidates.extend((0..10).map(|_| make_path_with_latency(None)));
1255 let result = prune_for_consistency(candidates, 8, 0);
1256 assert_eq!(result.len(), 8);
1257 let has_0_hop = result.iter().any(|p| p.total_latency_ms == Some(50));
1259 assert!(has_0_hop, "0-hop path with measured latency must survive pruning");
1260 }
1261
1262 #[test]
1263 fn prune_multi_hop_without_fundable_tickets_floor_is_unpopulated() {
1264 let candidates: Vec<_> = vec![
1267 make_path_with_tickets(Some(50), Some(1_000)), make_path_with_tickets(Some(50), Some(1_000)), make_path_with_tickets(Some(50), Some(1_000)), make_path_with_tickets(Some(50), Some(1_000)), make_path_with_tickets(Some(50), Some(1_000)), make_path_with_tickets(Some(50), Some(1_000)), make_path_with_tickets(Some(50), Some(1_000)), make_path_with_tickets(Some(50), Some(1_000)), make_path_with_tickets(Some(40), None), ];
1277 let result = prune_for_consistency(candidates, 8, 1);
1278 assert_eq!(result.len(), 8);
1279 let has_missing_balance = result.iter().any(|p| p.fundable_tickets_floor.is_none());
1282 assert!(
1283 !has_missing_balance,
1284 "path without capacity floor must be pruned when fully-measured paths fill the floor"
1285 );
1286 }
1287
1288 #[test]
1289 fn prune_for_consistency_floor_zero_returns_all() {
1290 let candidates = vec![
1292 make_path_with_tickets(Some(10), Some(1_000)),
1293 make_path_with_tickets(Some(20), None),
1294 make_path_with_tickets(None, None),
1295 ];
1296 let result = prune_for_consistency(candidates, 0, 1);
1297 assert_eq!(result.len(), 3, "floor=0 must return all candidates");
1298 }
1299
1300 #[tokio::test]
1303 async fn path_metrics_aggregate_latency_correctly() -> anyhow::Result<()> {
1304 let me = pubkey(&SECRET_0);
1307 let a = pubkey(&SECRET_1);
1308 let b = pubkey(&SECRET_2);
1309 let dest = pubkey(&SECRET_3);
1310 let graph = ChannelGraph::new(me);
1311 for n in [a, b, dest] {
1312 graph.add_node(n);
1313 }
1314
1315 let make_edge = |src: &OffchainPublicKey, dst: &OffchainPublicKey, lat_ms: u64| {
1316 graph.upsert_edge(src, dst, |obs| {
1317 obs.record(EdgeWeightType::Connected(true));
1318 obs.record(EdgeWeightType::Immediate(Ok(Duration::from_millis(lat_ms))));
1319 obs.record(EdgeWeightType::Balance(Some(hopr_api::graph::traits::Balance::from(
1320 1000u64,
1321 ))));
1322 });
1323 };
1324
1325 for _ in 0..20 {
1327 make_edge(&me, &a, 30);
1328 make_edge(&a, &b, 40);
1329 make_edge(&b, &dest, 50);
1330 }
1331
1332 graph.add_edge(&me, &a).unwrap();
1334 graph.add_edge(&a, &b).unwrap();
1335 graph.add_edge(&b, &dest).unwrap();
1336
1337 let selector = test_selector(me, graph, MAX_PATHS);
1338 let paths = selector.select_path(me, dest, 2).context("forward 2-hop path")?;
1339 assert!(!paths.is_empty());
1340
1341 let total = paths[0].total_latency_ms.expect("latency must be Some");
1342 assert!(
1343 (100..=130).contains(&total),
1344 "expected ~120ms total latency, got {total}ms"
1345 );
1346 Ok(())
1347 }
1348
1349 #[tokio::test]
1350 async fn path_metrics_fundable_tickets_floor_is_min() -> anyhow::Result<()> {
1351 let me = pubkey(&SECRET_0);
1352 let hop = pubkey(&SECRET_1);
1353 let dest = pubkey(&SECRET_2);
1354 let graph = ChannelGraph::new(me);
1355 graph.add_node(hop);
1356 graph.add_node(dest);
1357
1358 graph.set_ticket_face_value(hopr_api::graph::traits::Balance::one());
1362
1363 graph.upsert_edge(&me, &hop, |obs| {
1364 obs.record(EdgeWeightType::Connected(true));
1365 obs.record(EdgeWeightType::Intermediate(Ok(Duration::from_millis(50))));
1366 obs.record(EdgeWeightType::Balance(Some(hopr_api::graph::traits::Balance::from(
1367 500u64,
1368 ))));
1369 });
1370 graph.upsert_edge(&hop, &dest, |obs| {
1371 obs.record(EdgeWeightType::Connected(true));
1372 obs.record(EdgeWeightType::Intermediate(Ok(Duration::from_millis(50))));
1373 obs.record(EdgeWeightType::Balance(Some(hopr_api::graph::traits::Balance::from(
1374 200u64,
1375 ))));
1376 });
1377 graph.add_edge(&me, &hop).unwrap();
1378 graph.add_edge(&hop, &dest).unwrap();
1379
1380 let selector = test_selector(me, graph, MAX_PATHS);
1381 let paths = selector.select_path(me, dest, 1).context("1-hop path")?;
1382 assert!(!paths.is_empty());
1383 assert_eq!(
1384 paths[0].fundable_tickets_floor,
1385 Some(200),
1386 "floor must be the smaller of 500 and 200"
1387 );
1388 Ok(())
1389 }
1390
1391 }