hopr_transport/path/
planner.rs

1use std::{sync::Arc, time::Duration};
2
3use futures::{StreamExt as _, TryStreamExt, stream::FuturesUnordered};
4#[cfg(all(feature = "telemetry", not(test)))]
5use hopr_api::types::internal::path::Path;
6use hopr_api::{
7    OffchainPublicKey,
8    chain::{ChainKeyOperations, ChainPathResolver, ChainReadChannelOperations},
9    types::{
10        crypto::crypto_traits::Randomizable,
11        internal::{errors::PathError, prelude::*},
12        primitive::traits::ToHex,
13    },
14};
15use hopr_crypto_packet::prelude::*;
16use hopr_protocol_hopr::{FoundSurb, SurbStore};
17use tracing::trace;
18use validator::{Validate, ValidationError};
19
20use super::{
21    errors::{PathPlannerError, Result},
22    traits::{BackgroundPathCacheRefreshable, PathSelector, PathWithMetrics},
23};
24
25#[cfg(all(feature = "telemetry", not(test)))]
26lazy_static::lazy_static! {
27    static ref METRIC_PATH_LENGTH: hopr_api::types::telemetry::SimpleHistogram = hopr_api::types::telemetry::SimpleHistogram::new(
28        "hopr_path_length",
29        "Distribution of number of hops of sent messages",
30        vec![0.0, 1.0, 2.0, 3.0, 4.0]
31    ).unwrap();
32}
33
34/// Configuration for [`PathPlanner`]'s internal path cache.
35#[derive(Debug, Clone, Copy, PartialEq, smart_default::SmartDefault, Validate)]
36pub struct PathPlannerConfig {
37    /// Maximum number of `(source, destination, options)` entries in the path cache.
38    #[default = 10_000]
39    pub max_cache_capacity: u64,
40    /// Time-to-live for a cached path list.  When an entry expires the next
41    /// [`PathPlanner::resolve_routing`] call transparently recomputes it (lazy refresh).
42    #[default(Duration::from_secs(60))]
43    pub cache_ttl: Duration,
44    /// Period between proactive background cache-refresh sweeps.
45    #[default(Duration::from_secs(30))]
46    pub refresh_period: Duration,
47    /// Maximum number of candidate paths the selector may return per query.
48    /// All returned candidates are validated and cached.
49    #[default = 50]
50    pub max_cached_paths: usize,
51    /// Penalty multiplier for edges lacking probe-based quality observations.
52    /// Applied during path cost evaluation to down-weight unprobed edges.
53    /// Must be finite and in `0.0..=1.0`.
54    #[default = 0.5]
55    #[validate(custom(function = "validate_unit_interval"))]
56    pub edge_penalty: f64,
57    /// Minimum acceptable message acknowledgment rate for path selection.
58    /// Edges with an ack rate below this threshold are excluded from candidate paths.
59    /// Must be finite and in `0.0..=1.0`.
60    #[default = 0.1]
61    #[validate(custom(function = "validate_unit_interval"))]
62    pub min_ack_rate: f64,
63    /// Candidate count below which no latency-based pruning occurs.
64    ///
65    /// When fewer paths than this value are found, the selector returns all of them
66    /// unchanged (`min(found_count, floor)` semantics — the floor is never a minimum
67    /// to fabricate).  Set to 0 to disable pruning entirely.
68    #[default = 8]
69    pub min_paths_anonymity_floor: usize,
70    /// Total path latency at which the latency factor in the composite weight equals 0.5.
71    /// Higher values make the weight less sensitive to latency differences.
72    #[default(Duration::from_millis(100))]
73    pub latency_halflife: Duration,
74    /// Reference channel balance used to scale the capacity factor in the composite weight.
75    /// `capacity_factor` saturates at 1.0 near this value.
76    /// Defaults to 10_000_000 (~10 MiB in wxHOPR tokens).
77    #[default = 10_000_000]
78    pub capacity_reference: u128,
79    /// Upper bound on a loopback probe's round-trip time considered plausible.
80    /// Measurements above this cap (clock skew, stale telemetry) are discarded
81    /// instead of poisoning the latency EMA with an absurd value.
82    #[default(Duration::from_secs(30))]
83    pub max_plausible_loopback_rtt: Duration,
84}
85
86fn validate_unit_interval(value: f64) -> std::result::Result<(), ValidationError> {
87    if value.is_finite() && (0.0..=1.0).contains(&value) {
88        Ok(())
89    } else {
90        Err(ValidationError::new("value must be finite and in 0.0..=1.0"))
91    }
92}
93
94/// Parameters that shape how per-path aggregates modulate the `WeightedCollection` weight.
95#[derive(Debug, Clone, Copy)]
96struct WeightingParams {
97    latency_halflife: Duration,
98    capacity_reference: u128,
99}
100
101/// Continuous, monotonically decreasing latency factor, bounded in (0, 1].
102///
103/// Returns 1.0 for zero latency and 0.5 when `latency == halflife`.
104fn latency_factor(latency: Duration, halflife: Duration) -> f64 {
105    let ms = latency.as_millis() as f64;
106    let h = halflife.as_millis().max(1) as f64;
107    1.0 / (1.0 + ms / h)
108}
109
110/// Continuous, monotonically increasing capacity factor, bounded in (0.05, 1.0].
111///
112/// Uses a log scale because channel balances span many orders of magnitude.
113/// Saturates at 1.0 near `reference`.
114fn capacity_factor(c: u128, reference: u128) -> f64 {
115    let log = (c as f64).max(1.0).log10();
116    let ref_log = (reference as f64).max(10.0).log10();
117    (log / ref_log).clamp(0.05, 1.0)
118}
119
120/// Composite selection weight for a candidate path.
121///
122/// Refines `pwc.cost` (the `EdgeValueFn` output) with latency and capacity factors
123/// derived from the per-path aggregates.  Factors are neutral (1.0) when the
124/// corresponding aggregate is unavailable to avoid penalising unprobed paths.
125/// For 0-hop routes (`hops == 0`) the capacity factor is always 1.0 — direct
126/// `me -> dest` packets use no payment channel, so `capacity_floor = None` is expected.
127fn composite_weight(pwc: &PathWithMetrics, hops: usize, params: WeightingParams) -> f64 {
128    let lat = pwc
129        .total_latency_ms
130        .map(|ms| latency_factor(Duration::from_millis(ms as u64), params.latency_halflife))
131        .unwrap_or(1.0);
132    let cap = if hops == 0 {
133        1.0
134    } else {
135        pwc.capacity_floor
136            .map(|c| capacity_factor(c, params.capacity_reference))
137            .unwrap_or(1.0)
138    };
139    pwc.cost * lat * cap
140}
141
142/// Cache key for the path planner: `(source, destination, hops)`.
143///
144/// Only the `Hops` variant of [`RoutingOptions`] is cached (explicit intermediate
145/// paths bypass the cache), so the key stores the hop count as a plain `u32`.
146type PlannerCacheKey = (NodeId, NodeId, u32);
147type PlannerCacheValue = Arc<hopr_utils::statistics::WeightedCollection<ValidatedPath>>;
148
149/// Path planner that resolves [`DestinationRouting`] to [`ResolvedTransportRouting`].
150///
151/// The planner delegates path *discovery* to any [`PathSelector`] implementation and
152/// owns the `moka` cache of fully-validated [`ValidatedPath`] objects paired with
153/// their traversal cost, keyed by `(source: NodeId, destination: NodeId, hops: u32)`.
154///
155/// On a cache miss the planner calls the selector, validates every candidate against
156/// the chain resolver, and stores an `Arc<WeightedCollection<ValidatedPath>>` in the
157/// cache. On a cache hit a candidate is picked via weighted random selection (higher
158/// cost = higher quality = higher probability).
159///
160/// A background sweep (`background_refresh`) can be spawned to
161/// proactively re-warm the cache for all previously-seen keys.
162#[derive(Clone)]
163pub struct PathPlanner<Surb, R, S> {
164    me: OffchainPublicKey,
165    pub surb_store: Surb,
166    resolver: Arc<R>,
167    selector: Arc<S>,
168    cache: moka::future::Cache<PlannerCacheKey, PlannerCacheValue>,
169    refresh_period: Duration,
170    weighting: WeightingParams,
171}
172
173impl<Surb, R, S> PathPlanner<Surb, R, S>
174where
175    Surb: SurbStore + Send + Sync + 'static,
176    R: ChainKeyOperations + ChainReadChannelOperations + Send + Sync + 'static,
177    S: PathSelector + Send + Sync + 'static,
178{
179    /// Create a new path planner.
180    ///
181    /// `me` is this node's [`OffchainPublicKey`]; it is used as the source in path queries.
182    pub fn new(me: OffchainPublicKey, surb_store: Surb, resolver: R, selector: S, config: PathPlannerConfig) -> Self {
183        let cache = moka::future::Cache::builder()
184            .max_capacity(config.max_cache_capacity)
185            .time_to_live(config.cache_ttl)
186            .build();
187
188        Self {
189            me,
190            surb_store,
191            resolver: Arc::new(resolver),
192            selector: Arc::new(selector),
193            cache,
194            refresh_period: config.refresh_period,
195            weighting: WeightingParams {
196                latency_halflife: config.latency_halflife,
197                capacity_reference: config.capacity_reference,
198            },
199        }
200    }
201
202    /// Resolve a [`NodeId`] to an [`OffchainPublicKey`].
203    async fn resolve_node_id_to_offchain_key(&self, node_id: &NodeId) -> Result<OffchainPublicKey> {
204        match node_id {
205            NodeId::Offchain(key) => Ok(*key),
206            NodeId::Chain(addr) => {
207                let resolver = ChainPathResolver::from(&*self.resolver);
208                resolver
209                    .resolve_transport_address(addr)
210                    .await
211                    .map_err(|e| PathPlannerError::Other(anyhow::anyhow!("{e}")))?
212                    .ok_or_else(|| {
213                        PathPlannerError::Other(anyhow::anyhow!("no offchain key found for chain address {addr}"))
214                    })
215            }
216        }
217    }
218
219    #[tracing::instrument(level = "trace", skip(self))]
220    async fn resolve_path(
221        &self,
222        source: NodeId,
223        destination: NodeId,
224        options: RoutingOptions,
225    ) -> Result<ValidatedPath> {
226        let path = match options {
227            RoutingOptions::IntermediatePath(explicit_path) => {
228                tracing::debug!(
229                    direction = "loopback",
230                    ?source,
231                    ?destination,
232                    ?explicit_path,
233                    "resolving intermediate path"
234                );
235                let resolver = ChainPathResolver::from(&*self.resolver);
236                ValidatedPath::new(
237                    source,
238                    explicit_path
239                        .into_iter()
240                        .chain(std::iter::once(destination))
241                        .collect::<Vec<_>>(),
242                    &resolver,
243                )
244                .await?
245            }
246
247            RoutingOptions::Hops(hops) if u32::from(hops) == 0 => {
248                trace!(hops = 0, "resolving zero-hop direct path");
249                let resolver = ChainPathResolver::from(&*self.resolver);
250                ValidatedPath::new(source, vec![destination], &resolver).await?
251            }
252
253            RoutingOptions::Hops(hops) => {
254                let hops_usize: usize = hops.into();
255                trace!(hops = hops_usize, "resolving path via planner cache");
256
257                let src_key = self.resolve_node_id_to_offchain_key(&source).await?;
258                let dest_key = self.resolve_node_id_to_offchain_key(&destination).await?;
259
260                let cache_key: PlannerCacheKey = (source, destination, u32::from(hops));
261
262                let resolver = self.resolver.clone();
263                let selector = self.selector.clone();
264                let weighting = self.weighting;
265
266                let paths = self
267                    .cache
268                    .try_get_with(cache_key, async move {
269                        trace!(hops = hops_usize, "path cache miss, querying selector");
270                        let candidates = selector.select_path(src_key, dest_key, hops_usize)?;
271
272                        let chain_resolver = ChainPathResolver::from(&*resolver);
273                        let mut valid_paths: Vec<(ValidatedPath, f64)> = Vec::with_capacity(candidates.len());
274                        let mut path_metrics: Vec<PathWithMetrics> = Vec::with_capacity(candidates.len());
275                        for mut pwc in candidates {
276                            let path_nodes = std::mem::take(&mut pwc.path);
277                            let node_ids: Vec<NodeId> =
278                                path_nodes.into_iter().map(NodeId::Offchain).collect::<Vec<_>>();
279                            match ValidatedPath::new(source, node_ids, &chain_resolver).await {
280                                Ok(vp) => {
281                                    valid_paths.push((vp, composite_weight(&pwc, hops_usize, weighting)));
282                                    path_metrics.push(pwc);
283                                }
284                                Err(e) => tracing::debug!(error = %e, "path candidate failed validation"),
285                            }
286                        }
287
288                        if valid_paths.is_empty() {
289                            return Err(PathPlannerError::Path(PathError::PathNotFound(
290                                hops_usize,
291                                src_key.to_hex(),
292                                dest_key.to_hex(),
293                            )));
294                        }
295
296                        let weighted = hopr_utils::statistics::WeightedCollection::new(valid_paths);
297                        let total_wt = weighted.total_weight();
298                        for ((vp, w), pwm) in weighted.iter().zip(path_metrics.iter()) {
299                            tracing::debug!(
300                                %destination,
301                                hops = hops_usize,
302                                path = %vp,
303                                cost = pwm.cost,
304                                composite_weight = w,
305                                sampling_probability = if total_wt > 0.0 && *w > 0.0 { *w / total_wt } else { 0.0 },
306                                total_latency_ms = ?pwm.total_latency_ms,
307                                min_probe_success_rate = ?pwm.min_probe_success_rate,
308                                min_ack_rate = ?pwm.min_ack_rate,
309                                capacity_floor = ?pwm.capacity_floor,
310                                "weighted candidate path",
311                            );
312                        }
313                        Ok(Arc::new(weighted))
314                    })
315                    .await
316                    .map_err(PathPlannerError::CacheError)?;
317
318                paths.pick_one().ok_or_else(|| {
319                    PathPlannerError::Path(PathError::PathNotFound(hops_usize, src_key.to_hex(), dest_key.to_hex()))
320                })?
321            }
322        };
323
324        #[cfg(all(feature = "telemetry", not(test)))]
325        {
326            hopr_api::types::telemetry::SimpleHistogram::observe(&METRIC_PATH_LENGTH, (path.num_hops() - 1) as f64);
327        }
328
329        trace!(%path, "validated resolved path");
330        Ok(path)
331    }
332
333    /// Resolve a [`DestinationRouting`] to a [`ResolvedTransportRouting`].
334    ///
335    /// Returns the resolved routing and, for `Return` variants, the number of remaining SURBs.
336    #[tracing::instrument(level = "trace", skip(self))]
337    pub async fn resolve_routing(
338        &self,
339        size_hint: usize,
340        max_surbs: usize,
341        routing: DestinationRouting,
342    ) -> Result<(ResolvedTransportRouting<HoprSurb>, Option<usize>)> {
343        match routing {
344            DestinationRouting::Forward {
345                destination,
346                pseudonym,
347                forward_options,
348                return_options,
349            } => {
350                tracing::debug!(direction = "forward", %destination, "resolving forward path");
351
352                let forward_path = self
353                    .resolve_path(NodeId::Offchain(self.me), *destination, forward_options)
354                    .await?;
355                tracing::debug!(direction = "forward", %destination, path = %forward_path, "resolved path");
356
357                let return_paths = if let Some(return_options) = return_options {
358                    let num_possible_surbs = HoprPacket::max_surbs_with_message(size_hint).min(max_surbs);
359                    trace!(
360                        %destination,
361                        %num_possible_surbs,
362                        data_len = size_hint,
363                        max_surbs,
364                        "resolving packet return paths"
365                    );
366
367                    (0..num_possible_surbs)
368                        .map(|_| self.resolve_path(*destination, NodeId::Offchain(self.me), return_options.clone()))
369                        .collect::<FuturesUnordered<_>>()
370                        .try_collect::<Vec<ValidatedPath>>()
371                        .await?
372                        .into_iter()
373                        .enumerate()
374                        .inspect(|(i, rp)| {
375                            tracing::debug!(direction = "return", %destination, index = i, path = %rp, "resolved return path");
376                        })
377                        .map(|(_, rp)| rp)
378                        .collect()
379                } else {
380                    vec![]
381                };
382
383                trace!(%destination, num_surbs = return_paths.len(), data_len = size_hint, "resolved packet");
384
385                Ok((
386                    ResolvedTransportRouting::Forward {
387                        pseudonym: pseudonym.unwrap_or_else(HoprPseudonym::random),
388                        forward_path,
389                        return_paths,
390                    },
391                    None,
392                ))
393            }
394
395            DestinationRouting::Return(matcher) => {
396                let FoundSurb {
397                    sender_id,
398                    surb,
399                    remaining,
400                } = self
401                    .surb_store
402                    .find_surb(matcher)
403                    .ok_or_else(|| PathPlannerError::Surb(format!("no surb for pseudonym {}", matcher.pseudonym())))?;
404                Ok((ResolvedTransportRouting::Return(sender_id, surb), Some(remaining)))
405            }
406        }
407    }
408}
409
410impl<Surb, R, S> BackgroundPathCacheRefreshable for PathPlanner<Surb, R, S>
411where
412    Surb: SurbStore + Send + Sync + 'static,
413    R: ChainKeyOperations + ChainReadChannelOperations + Send + Sync + 'static,
414    S: PathSelector + Send + Sync + 'static,
415{
416    /// Returns a future that runs the background path-cache refresh loop.
417    ///
418    /// The returned future iterates over all keys currently in the planner's cache
419    /// and recomputes their paths on a configurable schedule, so that steady-state
420    /// traffic is always served from cache.
421    fn run_background_refresh(&self) -> impl std::future::Future<Output = ()> + Send + 'static {
422        // Clone only the fields we need — avoids requiring R: Clone + S: Clone.
423        let cache = self.cache.clone();
424        let resolver = self.resolver.clone();
425        let selector = self.selector.clone();
426        let refresh_period = self.refresh_period;
427        let weighting = self.weighting;
428
429        // run at a non-zero interval
430        futures_time::stream::interval(futures_time::time::Duration::from_millis(
431            refresh_period.as_millis() as u64 + 1u64,
432        ))
433        .for_each(move |_| {
434            let cache = cache.clone();
435            let resolver = resolver.clone();
436            let selector = selector.clone();
437            let weighting = weighting;
438
439            async move {
440                for (key, _) in cache.iter() {
441                    let (src, dest, hops_u32) = {
442                        let k = key.as_ref();
443                        (k.0, k.1, k.2)
444                    };
445
446                    if hops_u32 == 0 {
447                        continue;
448                    }
449                    let hops_usize = hops_u32 as usize;
450
451                    let resolve_key = |node: NodeId| {
452                        let resolver = resolver.clone();
453
454                        async move {
455                            match node {
456                                NodeId::Offchain(k) => Some(k),
457                                NodeId::Chain(addr) => ChainPathResolver::from(&*resolver)
458                                    .resolve_transport_address(&addr)
459                                    .await
460                                    .ok()
461                                    .flatten(),
462                            }
463                        }
464                    };
465
466                    if let (Some(src_key), Some(dest_key)) = (resolve_key(src).await, resolve_key(dest).await)
467                        && let Ok(candidates) = selector.select_path(src_key, dest_key, hops_usize)
468                    {
469                        let chain_resolver = ChainPathResolver::from(&*resolver);
470                        let mut valid_paths: Vec<(ValidatedPath, f64)> = Vec::with_capacity(candidates.len());
471                        let mut path_metrics: Vec<PathWithMetrics> = Vec::with_capacity(candidates.len());
472                        for mut pwc in candidates {
473                            let path_nodes = std::mem::take(&mut pwc.path);
474                            let node_ids: Vec<NodeId> =
475                                path_nodes.into_iter().map(NodeId::Offchain).collect::<Vec<_>>();
476                            match ValidatedPath::new(src, node_ids, &chain_resolver).await {
477                                Ok(vp) => {
478                                    valid_paths.push((vp, composite_weight(&pwc, hops_usize, weighting)));
479                                    path_metrics.push(pwc);
480                                }
481                                Err(e) => {
482                                    tracing::debug!(error = %e, "background refresh: path candidate failed validation")
483                                }
484                            }
485                        }
486
487                        if !valid_paths.is_empty() {
488                            let weighted = hopr_utils::statistics::WeightedCollection::new(valid_paths);
489                            let total_wt = weighted.total_weight();
490                            for ((vp, w), pwm) in weighted.iter().zip(path_metrics.iter()) {
491                                tracing::debug!(
492                                    kind = "background-refresh",
493                                    destination = %dest_key,
494                                    hops = hops_usize,
495                                    path = %vp,
496                                    cost = pwm.cost,
497                                    composite_weight = w,
498                                    sampling_probability = if total_wt > 0.0 && *w > 0.0 { *w / total_wt } else { 0.0 },
499                                    total_latency_ms = ?pwm.total_latency_ms,
500                                    min_probe_success_rate = ?pwm.min_probe_success_rate,
501                                    min_ack_rate = ?pwm.min_ack_rate,
502                                    capacity_floor = ?pwm.capacity_floor,
503                                    "weighted candidate path",
504                                );
505                            }
506                            cache.insert((src, dest, hops_u32), Arc::new(weighted)).await;
507                        }
508                    }
509                }
510            }
511        })
512    }
513}
514
515#[cfg(test)]
516mod tests {
517    use std::str::FromStr;
518
519    use bimap::BiMap;
520    use futures::stream::{self, BoxStream};
521    use hex_literal::hex;
522    use hopr_api::{
523        chain::{ChainKeyOperations, ChainReadChannelOperations, ChannelSelector, HoprKeyIdent},
524        graph::{NetworkGraphWrite, traits::EdgeObservableWrite},
525        types::{
526            crypto::prelude::{Keypair, OffchainKeypair},
527            internal::channels::{ChannelEntry, ChannelStatus, generate_channel_id},
528            primitive::prelude::*,
529        },
530    };
531    use hopr_network_graph::ChannelGraph;
532
533    use super::*;
534    use crate::path::selector::HoprGraphPathSelector;
535
536    #[derive(Debug)]
537    struct TestError(String);
538
539    impl std::fmt::Display for TestError {
540        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
541            f.write_str(&self.0)
542        }
543    }
544
545    impl std::error::Error for TestError {}
546
547    const SECRET_ME: [u8; 32] = hex!("60741b83b99e36aa0c1331578156e16b8e21166d01834abb6c64b103f885734d");
548    const SECRET_A: [u8; 32] = hex!("71bf1f42ebbfcd89c3e197a3fd7cda79b92499e509b6fefa0fe44d02821d146a");
549    const SECRET_DEST: [u8; 32] = hex!("c24bd833704dd2abdae3933fcc9962c2ac404f84132224c474147382d4db2299");
550
551    fn pubkey(secret: &[u8; 32]) -> OffchainPublicKey {
552        *OffchainKeypair::from_secret(secret).expect("valid secret").public()
553    }
554
555    #[derive(Clone)]
556    struct Mapper {
557        map: Arc<BiMap<OffchainPublicKey, HoprKeyIdent>>,
558    }
559
560    impl KeyIdMapping<HoprKeyIdent, OffchainPublicKey> for Mapper {
561        fn map_key_to_id(&self, key: &OffchainPublicKey) -> Option<HoprKeyIdent> {
562            self.map.get_by_left(key).copied()
563        }
564
565        fn map_id_to_public(&self, id: &HoprKeyIdent) -> Option<OffchainPublicKey> {
566            self.map.get_by_right(id).copied()
567        }
568
569        fn map_keys_to_ids(&self, keys: &[OffchainPublicKey]) -> Vec<Option<HoprKeyIdent>> {
570            keys.iter().map(|key| self.map_key_to_id(key)).collect()
571        }
572
573        fn map_ids_to_keys(&self, ids: &[HoprKeyIdent]) -> Vec<Option<OffchainPublicKey>> {
574            ids.iter().map(|id| self.map_id_to_public(id)).collect()
575        }
576    }
577
578    struct TestChainApi {
579        me: Address,
580        key_addr_map: BiMap<OffchainPublicKey, Address>,
581        channels: Vec<ChannelEntry>,
582        id_mapper: Mapper,
583    }
584
585    impl TestChainApi {
586        fn new(me_key: OffchainPublicKey, me_addr: Address, peers: Vec<(OffchainPublicKey, Address)>) -> Self {
587            let mut key_addr_map = BiMap::new();
588            let mut key_id_map: BiMap<OffchainPublicKey, HoprKeyIdent> = BiMap::new();
589            key_addr_map.insert(me_key, me_addr);
590            key_id_map.insert(me_key, 0u32.into());
591            for (i, (k, a)) in peers.iter().enumerate() {
592                key_addr_map.insert(*k, *a);
593                key_id_map.insert(*k, ((i + 1) as u32).into());
594            }
595            Self {
596                me: me_addr,
597                key_addr_map,
598                channels: vec![],
599                id_mapper: Mapper {
600                    map: Arc::new(key_id_map),
601                },
602            }
603        }
604
605        fn with_open_channel(mut self, src: Address, dst: Address) -> Self {
606            self.channels.push(
607                ChannelEntry::builder()
608                    .between(src, dst)
609                    .amount(100)
610                    .ticket_index(1)
611                    .status(ChannelStatus::Open)
612                    .epoch(1)
613                    .build()
614                    .unwrap(),
615            );
616            self
617        }
618    }
619
620    impl ChainKeyOperations for TestChainApi {
621        type Error = TestError;
622        type Mapper = Mapper;
623
624        fn chain_key_to_packet_key(
625            &self,
626            chain: &Address,
627        ) -> std::result::Result<Option<OffchainPublicKey>, TestError> {
628            Ok(self.key_addr_map.get_by_right(chain).copied())
629        }
630
631        fn packet_key_to_chain_key(
632            &self,
633            packet: &OffchainPublicKey,
634        ) -> std::result::Result<Option<Address>, TestError> {
635            Ok(self.key_addr_map.get_by_left(packet).copied())
636        }
637
638        fn key_id_mapper_ref(&self) -> &Self::Mapper {
639            &self.id_mapper
640        }
641    }
642
643    impl ChainReadChannelOperations for TestChainApi {
644        type Error = TestError;
645
646        fn me(&self) -> &Address {
647            &self.me
648        }
649
650        fn channel_by_id(&self, channel_id: &ChannelId) -> std::result::Result<Option<ChannelEntry>, TestError> {
651            Ok(self
652                .channels
653                .iter()
654                .find(|c| generate_channel_id(&c.source, &c.destination) == *channel_id)
655                .cloned())
656        }
657
658        fn stream_channels<'a>(
659            &'a self,
660            _selector: ChannelSelector,
661        ) -> std::result::Result<BoxStream<'a, ChannelEntry>, TestError> {
662            Ok(Box::pin(stream::iter(self.channels.clone())))
663        }
664    }
665
666    // ── address fixtures ──────────────────────────────────────────────────────
667    fn me_addr() -> Address {
668        Address::from_str("0x1000d5786d9e6799b3297da1ad55605b91e2c882").expect("valid addr")
669    }
670    fn a_addr() -> Address {
671        Address::from_str("0x200060ddced1e33c9647a71f4fc2cf4ed33e4a9d").expect("valid addr")
672    }
673    fn dest_addr() -> Address {
674        Address::from_str("0x30004105095c8c10f804109b4d1199a9ac40ed46").expect("valid addr")
675    }
676
677    // ── graph helpers ──────────────────────────────────────────────────────────
678    fn mark_edge_full(graph: &ChannelGraph, src: &OffchainPublicKey, dst: &OffchainPublicKey) {
679        use hopr_api::graph::traits::EdgeWeightType;
680        graph.upsert_edge(src, dst, |obs| {
681            obs.record(EdgeWeightType::Connected(true));
682            obs.record(EdgeWeightType::Immediate(Ok(std::time::Duration::from_millis(50))));
683            obs.record(EdgeWeightType::Intermediate(Ok(std::time::Duration::from_millis(50))));
684            obs.record(EdgeWeightType::Capacity(Some(1000)));
685        });
686    }
687
688    fn small_config() -> PathPlannerConfig {
689        PathPlannerConfig {
690            max_cache_capacity: 100,
691            cache_ttl: std::time::Duration::from_secs(60),
692            refresh_period: std::time::Duration::from_secs(60),
693            max_cached_paths: 2,
694            ..PathPlannerConfig::default()
695        }
696    }
697
698    // ── test: zero-hop path ───────────────────────────────────────────────────
699
700    #[tokio::test]
701    async fn zero_hop_path_should_bypass_selector() {
702        let me = pubkey(&SECRET_ME);
703        let dest = pubkey(&SECRET_DEST);
704
705        // Build empty graph (no edges) — selector would fail if called.
706        let graph = ChannelGraph::new(me);
707        let cfg = small_config();
708        let selector = HoprGraphPathSelector::new(
709            me,
710            graph,
711            cfg.max_cached_paths,
712            cfg.edge_penalty,
713            cfg.min_ack_rate,
714            cfg.min_paths_anonymity_floor,
715        );
716
717        let chain_api = TestChainApi::new(me, me_addr(), vec![(dest, dest_addr())]);
718        let surb_store = hopr_protocol_hopr::MemorySurbStore::default();
719
720        let planner = PathPlanner::new(me, surb_store, chain_api, selector, small_config());
721
722        let routing = DestinationRouting::Forward {
723            destination: Box::new(NodeId::Offchain(dest)),
724            pseudonym: None,
725            forward_options: RoutingOptions::Hops(0.try_into().expect("valid 0")),
726            return_options: None,
727        };
728
729        let result = planner.resolve_routing(100, 0, routing).await;
730        assert!(result.is_ok(), "zero-hop should succeed: {:?}", result.err());
731
732        let (resolved, rem) = result.unwrap();
733        assert!(rem.is_none());
734        if let ResolvedTransportRouting::Forward { forward_path, .. } = resolved {
735            assert_eq!(
736                forward_path.num_hops(),
737                1,
738                "zero-hop = 1 node in path (just destination)"
739            );
740        } else {
741            panic!("expected Forward routing");
742        }
743    }
744
745    // ── test: one-hop path via graph selector ─────────────────────────────────
746
747    #[tokio::test]
748    async fn one_hop_path_should_use_selector() {
749        let me = pubkey(&SECRET_ME);
750        let a = pubkey(&SECRET_A);
751        let dest = pubkey(&SECRET_DEST);
752
753        let graph = ChannelGraph::new(me);
754        graph.add_node(a);
755        graph.add_node(dest);
756        graph.add_edge(&me, &a).unwrap();
757        graph.add_edge(&a, &dest).unwrap();
758        mark_edge_full(&graph, &me, &a);
759        mark_edge_full(&graph, &a, &dest);
760
761        let cfg = small_config();
762        let selector = HoprGraphPathSelector::new(
763            me,
764            graph,
765            cfg.max_cached_paths,
766            cfg.edge_penalty,
767            cfg.min_ack_rate,
768            cfg.min_paths_anonymity_floor,
769        );
770
771        let chain_api = TestChainApi::new(me, me_addr(), vec![(a, a_addr()), (dest, dest_addr())])
772            .with_open_channel(me_addr(), a_addr())
773            .with_open_channel(a_addr(), dest_addr());
774
775        let surb_store = hopr_protocol_hopr::MemorySurbStore::default();
776        let planner = PathPlanner::new(me, surb_store, chain_api, selector, small_config());
777
778        let routing = DestinationRouting::Forward {
779            destination: Box::new(NodeId::Offchain(dest)),
780            pseudonym: None,
781            forward_options: RoutingOptions::Hops(1.try_into().expect("valid 1")),
782            return_options: None,
783        };
784
785        let result = planner.resolve_routing(100, 0, routing).await;
786        assert!(result.is_ok(), "1-hop routing should succeed: {:?}", result.err());
787
788        let (resolved, _) = result.unwrap();
789        if let ResolvedTransportRouting::Forward { forward_path, .. } = resolved {
790            assert_eq!(
791                forward_path.num_hops(),
792                2,
793                "1 intermediate hop means path has 2 nodes [a, dest]"
794            );
795        } else {
796            panic!("expected Forward routing");
797        }
798    }
799
800    #[tokio::test]
801    async fn explicit_intermediate_path_should_bypass_selector() {
802        let me = pubkey(&SECRET_ME);
803        let a = pubkey(&SECRET_A);
804        let dest = pubkey(&SECRET_DEST);
805
806        // Empty graph — selector would fail; explicit path should not use it.
807        let graph = ChannelGraph::new(me);
808        let cfg = small_config();
809        let selector = HoprGraphPathSelector::new(
810            me,
811            graph,
812            cfg.max_cached_paths,
813            cfg.edge_penalty,
814            cfg.min_ack_rate,
815            cfg.min_paths_anonymity_floor,
816        );
817
818        let chain_api = TestChainApi::new(me, me_addr(), vec![(a, a_addr()), (dest, dest_addr())])
819            .with_open_channel(me_addr(), a_addr())
820            .with_open_channel(a_addr(), dest_addr());
821
822        let surb_store = hopr_protocol_hopr::MemorySurbStore::default();
823        let planner = PathPlanner::new(me, surb_store, chain_api, selector, small_config());
824
825        use hopr_api::types::primitive::prelude::BoundedVec;
826        let intermediate_path = BoundedVec::try_from(vec![NodeId::Offchain(a)]).expect("valid");
827
828        let routing = DestinationRouting::Forward {
829            destination: Box::new(NodeId::Offchain(dest)),
830            pseudonym: None,
831            forward_options: RoutingOptions::IntermediatePath(intermediate_path),
832            return_options: None,
833        };
834
835        let result = planner.resolve_routing(100, 0, routing).await;
836        assert!(result.is_ok(), "explicit path should succeed: {:?}", result.err());
837
838        let (resolved, _) = result.unwrap();
839        if let ResolvedTransportRouting::Forward { forward_path, .. } = resolved {
840            assert_eq!(forward_path.num_hops(), 2, "one intermediate + destination = 2 hops");
841        } else {
842            panic!("expected Forward routing");
843        }
844    }
845
846    #[tokio::test]
847    async fn return_routing_without_surb_should_return_error() {
848        let me = pubkey(&SECRET_ME);
849        let graph = ChannelGraph::new(me);
850        let cfg = small_config();
851        let selector = HoprGraphPathSelector::new(
852            me,
853            graph,
854            cfg.max_cached_paths,
855            cfg.edge_penalty,
856            cfg.min_ack_rate,
857            cfg.min_paths_anonymity_floor,
858        );
859        let chain_api = TestChainApi::new(me, me_addr(), vec![]);
860        let surb_store = hopr_protocol_hopr::MemorySurbStore::default();
861
862        let planner = PathPlanner::new(me, surb_store, chain_api, selector, small_config());
863
864        use hopr_api::types::internal::routing::SurbMatcher;
865        let matcher = SurbMatcher::Pseudonym(HoprPseudonym::random());
866        let routing = DestinationRouting::Return(matcher);
867
868        let result = planner.resolve_routing(0, 0, routing).await;
869        assert!(result.is_err(), "should fail when no SURB exists");
870        assert!(
871            matches!(result.unwrap_err(), PathPlannerError::Surb(_)),
872            "error should be Surb variant"
873        );
874    }
875
876    // ── test: cache integration ───────────────────────────────────────────────
877
878    #[tokio::test]
879    async fn planner_cache_miss_should_populate_cache() {
880        let me = pubkey(&SECRET_ME);
881        let a = pubkey(&SECRET_A);
882        let dest = pubkey(&SECRET_DEST);
883
884        let graph = ChannelGraph::new(me);
885        graph.add_node(a);
886        graph.add_node(dest);
887        graph.add_edge(&me, &a).unwrap();
888        graph.add_edge(&a, &dest).unwrap();
889        mark_edge_full(&graph, &me, &a);
890        mark_edge_full(&graph, &a, &dest);
891
892        let cfg = small_config();
893        let selector = HoprGraphPathSelector::new(
894            me,
895            graph,
896            cfg.max_cached_paths,
897            cfg.edge_penalty,
898            cfg.min_ack_rate,
899            cfg.min_paths_anonymity_floor,
900        );
901        let chain_api = TestChainApi::new(me, me_addr(), vec![(a, a_addr()), (dest, dest_addr())])
902            .with_open_channel(me_addr(), a_addr())
903            .with_open_channel(a_addr(), dest_addr());
904        let surb_store = hopr_protocol_hopr::MemorySurbStore::default();
905        let planner = PathPlanner::new(me, surb_store, chain_api, selector, small_config());
906
907        let cache_key: PlannerCacheKey = (NodeId::Offchain(me), NodeId::Offchain(dest), 1);
908
909        assert!(
910            planner.cache.get(&cache_key).await.is_none(),
911            "cache should be empty before first call"
912        );
913
914        let routing = DestinationRouting::Forward {
915            destination: Box::new(NodeId::Offchain(dest)),
916            pseudonym: None,
917            forward_options: RoutingOptions::Hops(1.try_into().expect("valid 1")),
918            return_options: None,
919        };
920        planner.resolve_routing(100, 0, routing).await.expect("should succeed");
921
922        let cached = planner.cache.get(&cache_key).await;
923        assert!(cached.is_some(), "cache should be populated after first call");
924        let paths = cached.unwrap();
925        assert!(!paths.is_empty(), "cached paths must be non-empty");
926        let (first_path, first_cost) = paths.iter().next().expect("at least one cached path");
927        assert_eq!(first_path.num_hops(), 2, "path should have 2 hops [a, dest]");
928        assert!(*first_cost > 0.0, "cost should be positive");
929    }
930
931    #[tokio::test]
932    async fn planner_cache_hit_should_return_valid_path() {
933        let me = pubkey(&SECRET_ME);
934        let a = pubkey(&SECRET_A);
935        let dest = pubkey(&SECRET_DEST);
936
937        let graph = ChannelGraph::new(me);
938        graph.add_node(a);
939        graph.add_node(dest);
940        graph.add_edge(&me, &a).unwrap();
941        graph.add_edge(&a, &dest).unwrap();
942        mark_edge_full(&graph, &me, &a);
943        mark_edge_full(&graph, &a, &dest);
944
945        let cfg = small_config();
946        let selector = HoprGraphPathSelector::new(
947            me,
948            graph,
949            cfg.max_cached_paths,
950            cfg.edge_penalty,
951            cfg.min_ack_rate,
952            cfg.min_paths_anonymity_floor,
953        );
954        let chain_api = TestChainApi::new(me, me_addr(), vec![(a, a_addr()), (dest, dest_addr())])
955            .with_open_channel(me_addr(), a_addr())
956            .with_open_channel(a_addr(), dest_addr());
957        let surb_store = hopr_protocol_hopr::MemorySurbStore::default();
958        let planner = PathPlanner::new(me, surb_store, chain_api, selector, small_config());
959
960        let make_routing = || DestinationRouting::Forward {
961            destination: Box::new(NodeId::Offchain(dest)),
962            pseudonym: None,
963            forward_options: RoutingOptions::Hops(1.try_into().expect("valid 1")),
964            return_options: None,
965        };
966
967        let (r1, _) = planner.resolve_routing(100, 0, make_routing()).await.expect("call 1");
968        let (r2, _) = planner.resolve_routing(100, 0, make_routing()).await.expect("call 2");
969
970        let hops1 = if let ResolvedTransportRouting::Forward { forward_path, .. } = r1 {
971            forward_path.num_hops()
972        } else {
973            panic!("expected Forward");
974        };
975        let hops2 = if let ResolvedTransportRouting::Forward { forward_path, .. } = r2 {
976            forward_path.num_hops()
977        } else {
978            panic!("expected Forward");
979        };
980        assert_eq!(hops1, 2);
981        assert_eq!(hops2, 2);
982    }
983
984    #[tokio::test]
985    async fn background_refresh_should_produce_a_future() {
986        let me = pubkey(&SECRET_ME);
987        let graph = ChannelGraph::new(me);
988        let cfg = small_config();
989        let selector = HoprGraphPathSelector::new(
990            me,
991            graph,
992            cfg.max_cached_paths,
993            cfg.edge_penalty,
994            cfg.min_ack_rate,
995            cfg.min_paths_anonymity_floor,
996        );
997        let chain_api = TestChainApi::new(me, me_addr(), vec![]);
998        let surb_store = hopr_protocol_hopr::MemorySurbStore::default();
999
1000        let planner = PathPlanner::new(me, surb_store, chain_api, selector, small_config());
1001        // Just ensure it compiles and produces a future.
1002        let _future = planner.run_background_refresh();
1003    }
1004
1005    // ── composite weight helpers ──────────────────────────────────────────────
1006
1007    fn default_weighting() -> WeightingParams {
1008        WeightingParams {
1009            latency_halflife: Duration::from_millis(100),
1010            capacity_reference: 10_000_000,
1011        }
1012    }
1013
1014    fn make_pwm(cost: f64, latency_ms: Option<u32>, capacity_floor: Option<u128>) -> PathWithMetrics {
1015        PathWithMetrics {
1016            path: vec![],
1017            cost,
1018            total_latency_ms: latency_ms,
1019            min_probe_success_rate: None,
1020            min_ack_rate: None,
1021            capacity_floor,
1022        }
1023    }
1024
1025    #[test]
1026    fn latency_factor_is_monotonic_decreasing() {
1027        let halflife = Duration::from_millis(100);
1028        let f0 = latency_factor(Duration::ZERO, halflife);
1029        let f100 = latency_factor(Duration::from_millis(100), halflife);
1030        let f200 = latency_factor(Duration::from_millis(200), halflife);
1031        assert!(
1032            f0 > f100 && f100 > f200,
1033            "must be strictly decreasing: {f0} > {f100} > {f200}"
1034        );
1035        assert!(
1036            (f100 - 0.5).abs() < 1e-9,
1037            "at halflife factor should be 0.5, got {f100}"
1038        );
1039        assert!(f0 <= 1.0, "factor must never exceed 1.0, got {f0}");
1040    }
1041
1042    #[test]
1043    fn capacity_factor_is_monotonic_increasing() {
1044        let reference = 10_000_000u128;
1045        let f_low = capacity_factor(100, reference);
1046        let f_mid = capacity_factor(1_000_000, reference);
1047        let f_ref = capacity_factor(reference, reference);
1048        assert!(
1049            f_low < f_mid && f_mid <= f_ref,
1050            "must be non-decreasing: {f_low} < {f_mid} <= {f_ref}"
1051        );
1052        assert!(f_ref <= 1.0, "factor must not exceed 1.0 at reference, got {f_ref}");
1053        assert!(f_low >= 0.05, "minimum clamp is 0.05, got {f_low}");
1054    }
1055
1056    #[test]
1057    fn composite_weight_for_0_hop_skips_capacity_factor() {
1058        let params = default_weighting();
1059        let pwm = make_pwm(0.6, Some(100), None);
1060        let w = composite_weight(&pwm, 0, params);
1061        let expected = 0.6 * latency_factor(Duration::from_millis(100), params.latency_halflife);
1062        assert!(
1063            (w - expected).abs() < 1e-9,
1064            "0-hop weight should ignore capacity: {w} != {expected}"
1065        );
1066        assert!(w > 0.0, "0-hop weight must be positive");
1067    }
1068
1069    #[test]
1070    fn composite_weight_with_all_aggregates_is_below_cost() {
1071        let params = default_weighting();
1072        let pwm = make_pwm(0.8, Some(150), Some(5_000_000));
1073        let w = composite_weight(&pwm, 1, params);
1074        assert!(
1075            w < pwm.cost,
1076            "composite must be below raw cost when factors < 1.0: {w} >= {}",
1077            pwm.cost
1078        );
1079        assert!(w > 0.0, "composite weight must be positive");
1080    }
1081
1082    #[test]
1083    fn composite_weight_missing_capacity_on_multi_hop_neutral() {
1084        let params = default_weighting();
1085        let pwm_with = make_pwm(0.7, Some(80), Some(8_000_000));
1086        let pwm_without = make_pwm(0.7, Some(80), None);
1087        let w_with = composite_weight(&pwm_with, 2, params);
1088        let w_without = composite_weight(&pwm_without, 2, params);
1089        // Missing capacity is neutral (1.0), so without-capacity weight equals cost * lat only.
1090        let expected_without = 0.7 * latency_factor(Duration::from_millis(80), params.latency_halflife);
1091        assert!((w_without - expected_without).abs() < 1e-9);
1092        // With capacity the factor adds further reduction (capacity_factor < 1.0 here).
1093        assert!(w_with <= w_without, "known capacity should not increase the weight");
1094    }
1095}
hopr_transport/path/planner.rs

hopr_transport/path/
planner.rs
