Skip to main content

hopr_transport/path/
selector.rs

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/// Accumulated path cost and quality aggregates, folded edge-by-edge during DFS.
21///
22/// `PartialOrd` / `PartialEq` compare only the `cost` field so the DFS
23/// pruning threshold (`min_cost`) operates on the same scalar as before.
24#[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
58/// Returns the minimum of two `Option<T>` values, preferring `Some` over `None`.
59fn 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
66/// Wraps an `EdgeValueFn<f64, W>` as a `ValueFn` whose `Value` type carries
67/// both the cost and per-path quality aggregates.
68///
69/// All cost semantics are delegated to the inner `EdgeValueFn`; the wrapper
70/// only adds the aggregate fold on the same `&Weight` reference already
71/// available during DFS, so no extra graph lookups are needed.
72struct MetricsValueFn<W: EdgeObservableRead> {
73    inner: EdgeValueFn<f64, W>,
74    /// Face value used to express each edge's balance as a count of single-hop tickets.
75    ///
76    /// Read once per query rather than stored on any edge: deriving a count at query time is
77    /// safe, whereas deriving it at edge-update time is what made a price change stale the
78    /// whole graph.
79    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            // Probe rate: taking the min of immediate and intermediate guards against nodes that
125            // look good on direct probes but degrade under multi-hop load.
126            // `and_then`, not `map`: a stream with no observations contributes nothing to the min
127            // rather than contributing a zero it never measured.
128            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            // Expressed in single-hop tickets, not base units: the weighting factor below is a
140            // log ratio, and a base-unit balance would compress every realistic channel into a
141            // sliver of its output range.
142            //
143            // `zip`, not a fallback: an absent face value means the price is unknown, not that it
144            // is one. Dividing a base-unit balance by `1` yields a ticket count that saturates to
145            // `u128::MAX` and reads downstream as unlimited capacity — the opposite of what not
146            // knowing should imply. Contribute nothing to the floor until pricing arrives.
147            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
176/// Trim the candidate set to lower median latency and minimise variance while
177/// preserving an anonymity floor.
178///
179/// Behaviour:
180/// - If `candidates.len() <= floor`, returns all candidates unchanged (`min(found_count, floor)` semantics — the floor
181///   is never a minimum to fabricate).
182/// - Sorts candidates with a known `total_latency_ms` ascending.
183/// - Drops from the high-latency tail until the total count equals `floor`, or until no populated candidates remain.
184/// - If still over the floor with no populated candidates left, drops unpopulated candidates from the input-order tail.
185///
186/// A path is "fully measured" — and therefore preferred over unmeasured alternatives —
187/// when `total_latency_ms` is known AND either `hops == 0` (direct path, no channel
188/// expected) OR `fundable_tickets_floor` is also known.  This prevents 0-hop direct paths from
189/// being demoted simply because they carry no channel-capacity data.
190pub fn prune_for_consistency(candidates: Vec<PathWithMetrics>, floor: usize, hops: usize) -> Vec<PathWithMetrics> {
191    // floor == 0 means "no pruning" — caller opts out entirely.
192    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    // Sort populated ascending by latency (lowest first → drop from the end).
202    populated.sort_by_key(|p| p.total_latency_ms.unwrap_or(u32::MAX));
203
204    // Prefer measured paths: keep as many populated paths as fit within the floor,
205    // then fill the remaining slots with unpopulated paths.  This ensures that
206    // latency-measured candidates are never discarded when unprobed paths alone
207    // would satisfy the floor.
208    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
218/// Takes `n` candidates from `sorted`, preferring not-yet-represented first relayers.
219///
220/// Plain truncation keeps the `n` lowest-latency paths, which in a well-connected network are often
221/// the *same* few relayers reached by different routes; every downstream draw then inherits that
222/// concentration. Relative order is preserved, so each pass stays latency-ordered.
223fn 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    // First pass: one path per distinct first relayer, best (lowest-latency) first.
233    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    // Second pass: fill any remaining slots with the best of the rest.
246    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
263/// Number of distinct first relayers across `paths`.
264///
265/// The first relayer is the only node the destination gets to pick when it uses a SURB, so this is
266/// the count that decides whether the return-path degradation detector has a sibling to corroborate
267/// against: a value of 1 is the blind spot where a dead relayer is indistinguishable from a quiet
268/// peer. Paths with no hops (`path` empty) contribute nothing — a 0-hop return has no relayer.
269pub(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
277/// Compute candidate paths from `src` to `dest` through `graph`.
278///
279/// `length` is the number of edges to traverse (= intermediate hops + 1).
280/// `take` caps the number of candidate paths returned.
281/// The graph crate returns only the intermediate nodes (both `src` and `dest` stripped);
282/// this function re-appends `dest` so callers receive `([intermediates..., dest], cost)`.
283fn 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/// A lightweight graph-backed path selector.
312///
313/// Returns all candidate paths for a `(src, dest, hops)` query directly from
314/// the network graph — no caching is performed here.  The caller (typically
315/// [`crate::path::planner::PathPlanner`]) is responsible for caching, TTL management,
316/// background refresh, and final path selection.
317///
318/// Stores the planner's own identity (`me`) so that it can choose the
319/// appropriate cost function:
320/// - forward path (`src == me`): [`EdgeValueFn::forward`]
321/// - return path (`dest == me`): [`EdgeValueFn::returning`]
322#[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    /// Create a new selector.
343    ///
344    /// * `me` – the planner's own offchain public key, used to determine path direction.
345    /// * `graph` – the network graph to query.
346    /// * `max_paths` – maximum number of candidate paths to return per query.
347    /// * `edge_penalty` – penalty multiplier for edges lacking probe-based quality observations.
348    /// * `min_ack_rate` – minimum acceptable message acknowledgment rate for path selection.
349    /// * `anonymity_floor` – minimum candidate count below which no latency-based pruning occurs.
350    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    /// Extended forward path search: find shorter paths using
369    /// [`EdgeValueFn::forward_without_self_loopback`] and append `dest` to each one.
370    ///
371    /// This handles the case where the last edge (relay -> dest) has no graph edge
372    /// (e.g. no payment channel) but the path planner can still assume the last hop
373    /// is reachable. Paths already found by Phase 1 are excluded via `existing`.
374    ///
375    /// The cost from the shorter traversal is preserved as-is — the missing last
376    /// edge contributes a neutral `1.0` multiplier (no quality data available).
377    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        // The caller's snapshot, so both phases of one query cost against the same face value.
385        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                // The graph crate already strips `src`; `path` is [intermediates…, terminator].
407                // Guard: if dest already appears as an intermediate or terminator, appending it
408                // would produce a non-adjacent duplicate that ValidatedPath::new rejects — skip early.
409                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    /// Return all candidate paths from `src` to `dest` via `hops` relays.
438    ///
439    /// Each returned tuple contains a path `Vec<OffchainPublicKey>` of length
440    /// `hops + 1` (`[intermediates..., dest]`; `src` excluded) paired with its
441    /// accumulated traversal cost and per-path quality aggregates.
442    ///
443    /// Returns `Err(PathNotFound)` when the graph yields no positive-cost paths.
444    ///
445    /// The function has a potential to run expensive operations, it should be benchmarked
446    /// in a production environment and possibly guarded (e.g. by offloading the long execution
447    /// in an async executor to avoid blocking the caller).
448    #[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        // One read for the whole query. A concurrent `set_ticket_face_value` between two reads
462        // would let the traversal cost and the fundable-ticket floor come from different snapshots
463        // of the same graph, so `composite_weight` would rank a candidate against itself.
464        let ticket_face_value = self.graph.ticket_face_value();
465
466        let paths = if src == self.me {
467            // Phase 1: search for full-length forward paths to dest.
468            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            // Phase 2: if not enough paths, do an extended search with EdgeValueFn::forward_without_self_loopback
487            // for (length - 1) edges and assume the last hop can be done by anybody.
488            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    /// Mark an edge as fully ready for intermediate routing.
609    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    // Helper: build a bidirectional 2-hop graph: me ↔ hop ↔ dest.
621    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        // Forward: me → hop → dest
629        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        // Reverse: dest → hop → me
634        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        // No edges at all — neither direction has a path.
647        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        // Bidirectional: me ↔ A ↔ B ↔ dest  (2 intermediate hops each way)
691        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        // Forward: me → A → B → dest
700        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        // Reverse: dest → B → A → me
707        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        // Bidirectional: me ↔ relay ↔ dest
736        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        // Bidirectional diamond: me ↔ a ↔ dest and me ↔ b ↔ dest
762        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        // Forward: me → a → dest,  me → b → dest
771        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        // Reverse: dest → a → me,  dest → b → me
780        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        // Graph has edges in both directions but no observations → all costs zero → pruned.
809        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        // No observations → cost function will return non-positive cost → pruned.
816
817        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        // Graph has direct edges both ways; requesting 2 hops should fail in both directions.
832        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        // In the real network, the last edge (relay → dest) may not have a
856        // payment channel. The graph has me → relay but NO relay → dest edge.
857        // The virtual last hop fallback should still find the forward path.
858        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        // Forward: me → relay (fully observed). NO relay → dest edge.
865        graph.add_edge(&me, &relay).unwrap();
866        mark_edge_full(&graph, &me, &relay);
867        // Reverse: dest → relay → me (both edges exist for the return path).
868        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        // Forward path should work via virtual last hop
876        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        // Return path should work normally (both edges exist)
887        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        // Bidirectional: me ↔ a ↔ b ↔ c ↔ dest  (3 intermediate hops each way)
900        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        // Forward: me → a → b → c → dest
910        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        // Reverse: dest → c → b → a → me
919        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        // If me has a direct edge to dest (e.g. an edge node with a channel to its
956        // exit server), Phase 2 must not emit [dest, dest] — appending dest to a
957        // candidate that already ends in dest forms a loop that ValidatedPath::new
958        // would catch anyway, but we want the guard to skip it cleanly.
959        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        // me → dest: direct channel (this is what the own_chain_addr fix exposes)
966        graph.add_edge(&me, &dest).unwrap();
967        mark_edge_full(&graph, &me, &dest);
968        // me → relay: for 1-hop via relay
969        graph.add_edge(&me, &relay).unwrap();
970        mark_edge_full(&graph, &me, &relay);
971        // Return path
972        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        // Same as above but without relay→dest edge — only Phase 2 (virtual last hop)
994        // is in play. The guard must skip the me→dest edge as a relay candidate.
995        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        // me → dest: direct channel (no relay→dest edge)
1002        graph.add_edge(&me, &dest).unwrap();
1003        mark_edge_full(&graph, &me, &dest);
1004        // me → relay (relay is a valid intermediate; no relay→dest edge needed for Phase 2)
1005        graph.add_edge(&me, &relay).unwrap();
1006        mark_edge_full(&graph, &me, &relay);
1007        // Return path
1008        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        // Build graph with edges but NO observations → cost function returns 0.
1029        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        // No mark_edge_full/mark_edge_last → observations are empty → cost = 0
1038
1039        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    // ── pruning tests ─────────────────────────────────────────────────────────
1052
1053    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    /// A candidate whose first hop is relayer `relayer_idx`, with the given latency.
1076    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        // The four lowest-latency paths all leave via relayer 1; relayers 2..4 are slower. A pure
1093        // latency truncation to 3 would keep only relayer 1, collapsing all downstream diversity.
1094        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        // The best path per relayer is the one kept, so latency still drives the choice within
1117        // each relayer.
1118        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        // Three paths, two of them via the same relayer → two distinct first relayers. Paths with
1124        // no hops contribute nothing.
1125        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)), // empty path
1130        ];
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        // Ten qualifying paths, each via a distinct relayer. This is the "if 10 qualify, use 10"
1137        // case: with the cap disabled (floor = 0) every relayer survives, so the return-path
1138        // degradation detector always has siblings to corroborate against. A small cap (2) throws
1139        // eight good relayers away — the collapse that made the 2026-08-28 outage undetectable.
1140        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        // Only two relayers available but a floor of 4: the remaining slots go to the next-best
1154        // paths regardless of relayer.
1155        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        // 30 paths with strictly increasing latency, floor=8 → keep lowest 8
1181        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        // 3 populated + 6 unpopulated, floor=8
1194        // total=9 > floor=8: all 3 populated are kept (populated always preferred),
1195        // then 5 unpopulated fill the remaining slots.
1196        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        // All 3 populated paths survive; 1 unpopulated is trimmed.
1205        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        // 0 populated, 20 unpopulated, floor=8 → keep first 8
1215        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        // Regression: 2 populated + 10 unpopulated, floor=8.
1223        // Old formula: target_populated = 8.saturating_sub(10) = 0 → both populated dropped.
1224        // Correct: keep up to 8 populated (only 2 exist), fill 6 remaining with unpopulated.
1225        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        // 0-hop: fundable_tickets_floor = None is expected; path should be treated as "fully measured"
1250        // if latency is known.
1251        let mut candidates: Vec<_> = vec![
1252            make_path_with_tickets(Some(50), None), // 0-hop: no capacity, latency known
1253        ];
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        // The 0-hop path must be in the populated bucket and survive.
1258        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        // A 1-hop path with measured latency but NO capacity is unmeasured (unpopulated).
1265        // It should be demoted below paths that have both latency and capacity.
1266        let candidates: Vec<_> = vec![
1267            make_path_with_tickets(Some(50), Some(1_000)), // fully measured
1268            make_path_with_tickets(Some(50), Some(1_000)), // fully measured
1269            make_path_with_tickets(Some(50), Some(1_000)), // fully measured
1270            make_path_with_tickets(Some(50), Some(1_000)), // fully measured
1271            make_path_with_tickets(Some(50), Some(1_000)), // fully measured
1272            make_path_with_tickets(Some(50), Some(1_000)), // fully measured
1273            make_path_with_tickets(Some(50), Some(1_000)), // fully measured
1274            make_path_with_tickets(Some(50), Some(1_000)), // fully measured
1275            make_path_with_tickets(Some(40), None),        // missing capacity → unpopulated
1276        ];
1277        let result = prune_for_consistency(candidates, 8, 1);
1278        assert_eq!(result.len(), 8);
1279        // The path with missing capacity should not survive when all 8 slots are filled
1280        // by fully measured paths.
1281        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        // floor == 0 must be treated as "no pruning" — all candidates survive unchanged.
1291        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    // ── path metrics aggregation tests ────────────────────────────────────────
1301
1302    #[tokio::test]
1303    async fn path_metrics_aggregate_latency_correctly() -> anyhow::Result<()> {
1304        // 3-hop path: me → A (30ms) → B (40ms) → dest (50ms)
1305        // Total latency accumulated during DFS should be ~120ms.
1306        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        // Drive EMA to convergence with many samples at the target value.
1326        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        // Edges only in the forward direction (me → a → b → dest)
1333        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        // Stated, not assumed. A face value of one makes the ticket count equal the balance, which
1359        // is what keeps the numbers below readable — but it has to be pushed, because without a
1360        // price the floor is now `None` rather than silently dividing by one.
1361        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    // NOTE: a test for fundable_tickets_floor=None is omitted intentionally.
1392    // The forward cost function requires a funding balance on all non-last edges, so
1393    // every path the selector returns has capacity data on at least one edge, making
1394    // fundable_tickets_floor always Some for reachable paths.
1395}