hopr_lib/
lib.rs

1//! HOPR library creating a unified [`Hopr`] object that can be used on its own,
2//! as well as integrated into other systems and libraries.
3//!
4//! The [`Hopr`] object is standalone, meaning that once it is constructed and run,
5//! it will perform its functionality autonomously. The API it offers serves as a
6//! high-level integration point for other applications and utilities, but offers
7//! a complete and fully featured HOPR node stripped from top level functionality
8//! such as the REST API, key management...
9//!
10//! The intended way to use hopr_lib is for a specific tool to be built on top of it;
11//! should the default `hoprd` implementation not be acceptable.
12//!
13//! For most of the practical use cases, the `hoprd` application should be a preferable
14//! choice.
15/// Helper functions.
16mod helpers;
17
18/// Builder module for the [`Hopr`] object.
19pub mod builder;
20/// Configuration-related public types
21pub mod config;
22/// Various public constants.
23pub mod constants;
24/// Lists all errors thrown from this library.
25pub mod errors;
26/// Testing utilities: cluster fixtures, node wiring helpers, echo server.
27#[cfg(any(feature = "testing", test))]
28pub mod testing;
29/// Utility module with helper types and functionality over hopr-lib behavior.
30pub mod utils;
31
32pub use hopr_api as api;
33
34/// Exports of libraries necessary for API and interface operations.
35///
36/// Use `hopr_lib::api::types::*` for all type access.
37/// This module retains transport and network-specific types not available in `hopr_lib::api`.
38#[doc(hidden)]
39pub mod exports {
40    pub mod network {
41        pub use hopr_utils::network_types as types;
42    }
43
44    pub use hopr_transport as transport;
45}
46
47use std::{
48    sync::{Arc, atomic::Ordering},
49    time::Duration,
50};
51
52use futures::{FutureExt, Stream, StreamExt, TryFutureExt, pin_mut};
53use futures_concurrency::stream::Merge as _;
54use futures_time::future::FutureExt as FuturesTimeFutureExt;
55pub use hopr_api::types::keypair::key_pair::{HoprKeys, IdentityRetrievalModes};
56use hopr_api::{
57    PeerId,
58    chain::*,
59    graph::HoprGraphApi,
60    network::{Health, NetworkStreamControl, NetworkView},
61    node::{
62        ActionableEvent, ActionableEventDiscriminant, AtomicHoprState, ComponentStatus, ComponentStatusReporter,
63        EitherErrExt, EventWaitResult, HasChainApi, HasGraphView, HasNetworkView, HasTicketManagement, HasTransportApi,
64        HoprNodeOperations, HoprState, NodeOnchainIdentity,
65    },
66    tickets::TicketManagement,
67    types::{crypto::prelude::OffchainKeypair, internal::routing::DestinationRouting},
68};
69/// Maximum user-data payload per HOPR session frame (bytes).
70///
71/// Use this when sizing buffers or computing how many session frames a given
72/// wxHOPR balance can fund (together with the on-chain ticket price).
73pub use hopr_transport::SESSION_MTU;
74use hopr_transport::{ApplicationDataIn, ApplicationDataOut, HoprTransport, HoprTransportProcess, OffchainPublicKey};
75#[cfg(feature = "session-client")]
76pub use hopr_transport::{
77    FlowControlConfig, HoprSession, HoprSessionConfigurator, SessionCapabilities, SessionCapability, SessionTarget,
78    SurbBalancerConfig,
79};
80use hopr_utils::runtime::prelude::spawn;
81pub use hopr_utils::runtime::{Abortable, AbortableList};
82use tracing::debug;
83
84pub use crate::constants::{MIN_NATIVE_BALANCE, SUGGESTED_NATIVE_BALANCE};
85use crate::errors::HoprLibError;
86
87/// Public routing configuration for session opening in `hopr-lib`.
88///
89/// This intentionally exposes only hop-count based routing.
90#[cfg(feature = "session-client")]
91#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, smart_default::SmartDefault)]
92#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
93pub struct HopRouting(
94    #[default(hopr_api::types::primitive::bounded::BoundedSize::MIN)]
95    hopr_api::types::primitive::bounded::BoundedSize<
96        { hopr_api::types::internal::routing::RoutingOptions::MAX_INTERMEDIATE_HOPS },
97    >,
98);
99
100#[cfg(feature = "session-client")]
101impl HopRouting {
102    /// Maximum number of hops that can be configured.
103    pub const MAX_HOPS: usize = hopr_api::types::internal::routing::RoutingOptions::MAX_INTERMEDIATE_HOPS;
104
105    /// Returns the configured number of hops.
106    pub fn hop_count(self) -> usize {
107        self.0.into()
108    }
109}
110
111#[cfg(feature = "session-client")]
112impl TryFrom<usize> for HopRouting {
113    type Error = hopr_api::types::primitive::errors::GeneralError;
114
115    fn try_from(value: usize) -> Result<Self, Self::Error> {
116        Ok(Self(value.try_into()?))
117    }
118}
119
120#[cfg(feature = "session-client")]
121impl From<HopRouting> for hopr_api::types::internal::routing::RoutingOptions {
122    fn from(value: HopRouting) -> Self {
123        Self::Hops(value.0)
124    }
125}
126
127#[cfg(feature = "session-client")]
128impl std::fmt::Display for HopRouting {
129    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
130        write!(f, "{}-hop routing", self.hop_count())
131    }
132}
133
134/// Session client configuration for `hopr-lib`.
135///
136/// Unlike transport-level configuration, this API intentionally does not expose
137/// explicit intermediate paths.
138#[cfg(feature = "session-client")]
139#[derive(Debug, Clone, PartialEq, smart_default::SmartDefault)]
140pub struct HoprSessionClientConfig {
141    /// Forward route selection policy.
142    pub forward_path: HopRouting,
143    /// Return route selection policy.
144    pub return_path: HopRouting,
145    /// Capabilities offered by the session.
146    #[default(_code = "SessionCapability::Segmentation.into()")]
147    pub capabilities: SessionCapabilities,
148    /// Optional pseudonym used for the session. Mostly useful for testing only.
149    #[default(None)]
150    pub pseudonym: Option<hopr_api::types::internal::protocol::HoprPseudonym>,
151    /// Enable automatic SURB management for the session.
152    #[default(Some(SurbBalancerConfig::default()))]
153    pub surb_management: Option<SurbBalancerConfig>,
154    /// If set, the maximum number of possible SURBs will always be sent with session data packets.
155    #[default(false)]
156    pub always_max_out_surbs: bool,
157    /// Opt-in client-side send-window flow control for this session (`None` = unpaced, the default).
158    /// `Some(FlowControlConfig::default())` = the clean profile; `Some(FlowControlConfig::robust())` =
159    /// the tail-tolerance bundle. Only meaningful on a reliable (`RetransmissionAck`) session.
160    #[default(None)]
161    pub flow_control: Option<FlowControlConfig>,
162    /// Abandon the frame due next once the sequence has advanced this far past it, instead of
163    /// holding everything already received for the whole frame timeout.
164    ///
165    /// Head-of-line bound for this session's incoming direction. `None` inherits the node's
166    /// setting, `Some(0)` disables it here, `Some(n)` sets it.
167    ///
168    /// Worth setting per session because the right value tracks reordering depth -- throughput x
169    /// latency spread / frame size -- which is a property of the traffic, not of the node: a bulk
170    /// data session and a control session on the same node differ by orders of magnitude.
171    ///
172    /// Has no effect on a session carrying a retransmission capability, where a missing frame can
173    /// still be recovered and waiting for it is productive.
174    pub max_frames_behind_gap: Option<usize>,
175}
176
177/// Session client configuration for explicit intermediate-path routing.
178#[cfg(all(feature = "session-client", feature = "explicit-path"))]
179#[deprecated(
180    since = "4.0.2-rc.1",
181    note = "temporary compatibility API; remove once the explicit path is not needed anymore."
182)]
183#[derive(Debug, Clone, PartialEq)]
184pub struct HoprSessionClientExplicitPathConfig {
185    /// Explicit forward intermediate path.
186    pub forward_path: Vec<hopr_api::types::internal::NodeId>,
187    /// Explicit return intermediate path.
188    pub return_path: Vec<hopr_api::types::internal::NodeId>,
189    /// Capabilities offered by the session.
190    pub capabilities: SessionCapabilities,
191    /// Optional pseudonym used for the session. Mostly useful for testing only.
192    pub pseudonym: Option<hopr_api::types::internal::protocol::HoprPseudonym>,
193    /// Enable automatic SURB management for the session.
194    pub surb_management: Option<SurbBalancerConfig>,
195    /// If set, the maximum number of possible SURBs will always be sent with session data packets.
196    pub always_max_out_surbs: bool,
197    /// Opt-in client-side send-window flow control for this session (`None` = unpaced).
198    pub flow_control: Option<FlowControlConfig>,
199    /// As [`HoprSessionClientConfig::max_frames_behind_gap`].
200    pub max_frames_behind_gap: Option<usize>,
201}
202
203#[cfg(all(feature = "session-client", feature = "explicit-path"))]
204#[allow(deprecated)]
205impl Default for HoprSessionClientExplicitPathConfig {
206    fn default() -> Self {
207        Self {
208            forward_path: Vec::default(),
209            return_path: Vec::default(),
210            capabilities: SessionCapability::Segmentation.into(),
211            pseudonym: None,
212            surb_management: Some(SurbBalancerConfig::default()),
213            always_max_out_surbs: false,
214            flow_control: None,
215            max_frames_behind_gap: None,
216        }
217    }
218}
219
220#[cfg(feature = "session-client")]
221impl From<HoprSessionClientConfig> for hopr_transport::SessionClientConfig {
222    fn from(value: HoprSessionClientConfig) -> Self {
223        Self {
224            forward_path_options: value.forward_path.into(),
225            return_path_options: value.return_path.into(),
226            capabilities: value.capabilities,
227            pseudonym: value.pseudonym,
228            surb_management: value.surb_management,
229            always_max_out_surbs: value.always_max_out_surbs,
230            flow_control: value.flow_control,
231            max_frames_behind_gap: value.max_frames_behind_gap,
232        }
233    }
234}
235
236#[cfg(all(feature = "session-client", feature = "explicit-path"))]
237#[allow(deprecated)]
238impl TryFrom<HoprSessionClientExplicitPathConfig> for hopr_transport::SessionClientConfig {
239    type Error = hopr_api::types::primitive::errors::GeneralError;
240
241    fn try_from(value: HoprSessionClientExplicitPathConfig) -> Result<Self, Self::Error> {
242        let forward =
243            hopr_api::types::internal::routing::RoutingOptions::IntermediatePath(value.forward_path.try_into()?);
244        let ret = hopr_api::types::internal::routing::RoutingOptions::IntermediatePath(value.return_path.try_into()?);
245
246        Ok(Self {
247            forward_path_options: forward,
248            return_path_options: ret,
249            capabilities: value.capabilities,
250            pseudonym: value.pseudonym,
251            surb_management: value.surb_management,
252            always_max_out_surbs: value.always_max_out_surbs,
253            flow_control: value.flow_control,
254            max_frames_behind_gap: value.max_frames_behind_gap,
255        })
256    }
257}
258
259/// Long-running tasks that are spawned by the HOPR node.
260#[derive(Debug, Clone, PartialEq, Eq, Hash, strum::Display, strum::EnumCount)]
261pub(crate) enum HoprLibProcess {
262    #[strum(to_string = "transport: {0}")]
263    Transport(HoprTransportProcess),
264    #[strum(to_string = "session server providing the exit node session stream functionality")]
265    #[allow(dead_code)] // constructed only with feature = "session-server"
266    SessionServer,
267    #[strum(to_string = "subscription for on-chain channel updates")]
268    ChannelEvents,
269    #[strum(to_string = "on received ticket event (winning or rejected)")]
270    TicketEvents,
271    #[strum(to_string = "neglecting tickets on closed channels")]
272    ChannelClosureNeglect,
273}
274
275/// Prepare an optimized version of the tokio runtime setup for hopr-lib specifically.
276///
277/// Divide the available CPU parallelism by 2, since half of the available threads are
278/// to be used for IO-bound and half for CPU-bound tasks.
279#[cfg(feature = "runtime-tokio")]
280pub fn prepare_tokio_runtime(
281    num_cpu_threads: Option<std::num::NonZeroUsize>,
282    num_io_threads: Option<std::num::NonZeroUsize>,
283    thread_stack_size: Option<usize>,
284) -> anyhow::Result<tokio::runtime::Runtime> {
285    let avail_parallelism = std::thread::available_parallelism().ok().map(|v| v.get() / 2);
286
287    hopr_utils::parallelize::cpu::init_thread_pool(
288        num_cpu_threads
289            .map(|v| v.get())
290            .or(avail_parallelism)
291            .ok_or(anyhow::anyhow!(
292                "Could not determine the number of CPU threads to use. Please set the HOPRD_NUM_CPU_THREADS \
293                 environment variable."
294            ))?
295            .max(1),
296    )?;
297
298    Ok(tokio::runtime::Builder::new_multi_thread()
299        .enable_all()
300        .worker_threads(
301            num_io_threads
302                .map(|v| v.get())
303                .or(avail_parallelism)
304                .ok_or(anyhow::anyhow!(
305                    "Could not determine the number of IO threads to use. Please set the HOPRD_NUM_IO_THREADS \
306                     environment variable."
307                ))?
308                .max(1),
309        )
310        .thread_name("hoprd")
311        .thread_stack_size(thread_stack_size.unwrap_or(10 * 1024 * 1024).max(2 * 1024 * 1024))
312        .build()?)
313}
314
315/// Type alias used to send and receive transport data via a running HOPR node.
316pub type HoprTransportIO = hopr_transport::socket::HoprSocket<
317    futures::stream::BoxStream<'static, ApplicationDataIn>,
318    hopr_utils::network_types::crossfire_sink::CrossfireSink<(DestinationRouting, ApplicationDataOut)>,
319>;
320
321type TicketEvents = (
322    async_broadcast::Sender<hopr_api::node::TicketEvent>,
323    async_broadcast::InactiveReceiver<hopr_api::node::TicketEvent>,
324);
325
326/// Time to wait until the node's keybinding appears on-chain
327const NODE_READY_TIMEOUT: Duration = Duration::from_secs(120);
328
329/// HOPR main object providing the entire HOPR node functionality
330///
331/// Instantiating this object creates all processes and objects necessary for
332/// running the HOPR node. Once created, the node can be started using the
333/// `run()` method.
334///
335/// Externally offered API should be enough to perform all necessary tasks
336/// with the HOPR node manually, but it is advised to create such a configuration
337/// that manual interaction is unnecessary.
338///
339/// As such, the `hopr_lib` serves mainly as an integration point into Rust programs.
340pub struct Hopr<Chain, Graph, Net, TMgr> {
341    pub(crate) transport_id: OffchainKeypair,
342    pub(crate) chain_id: NodeOnchainIdentity,
343    pub(crate) cfg: config::HoprLibConfig,
344    pub(crate) state: Arc<AtomicHoprState>,
345    pub(crate) transport_api: HoprTransport<Chain, Graph, Net>,
346    pub(crate) chain_api: Chain,
347    pub(crate) ticket_event_subscribers: TicketEvents,
348    pub(crate) ticket_manager: TMgr,
349    #[allow(dead_code)] // Handles must stay alive to keep background tasks running
350    pub(crate) processes: AbortableList<HoprLibProcess>,
351}
352
353impl<Chain, Graph, Net, TMgr> std::fmt::Debug for Hopr<Chain, Graph, Net, TMgr> {
354    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
355        f.debug_struct("Hopr")
356            .field("identity", &self.chain_id)
357            .field("state", &self.state.load(std::sync::atomic::Ordering::Relaxed))
358            .field("config", &self.cfg)
359            .field("processes", &self.processes)
360            .finish_non_exhaustive()
361    }
362}
363
364impl<Chain, Graph, Net, TMgr> Hopr<Chain, Graph, Net, TMgr>
365where
366    Chain: HoprChainApi + Clone + Send + Sync + 'static,
367    Graph: HoprGraphApi<HoprNodeId = OffchainPublicKey> + Clone + Send + Sync + 'static,
368    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed:
369        hopr_api::graph::traits::EdgeObservableRead + Send + 'static,
370    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
371    Net: NetworkView + NetworkStreamControl + Send + Sync + Clone + 'static,
372{
373    pub fn config(&self) -> &config::HoprLibConfig {
374        &self.cfg
375    }
376
377    /// Returns a reference to the network graph.
378    pub fn graph(&self) -> &Graph {
379        self.transport_api.graph()
380    }
381
382    #[cfg(feature = "session-client")]
383    fn error_if_not_in_state(&self, state: HoprState, error: String) -> errors::Result<()> {
384        if HoprNodeOperations::status(self) == state {
385            Ok(())
386        } else {
387            Err(HoprLibError::NotReady(state, error))
388        }
389    }
390
391    #[cfg(feature = "session-client")]
392    async fn connect_to_with_transport_config(
393        &self,
394        destination: hopr_api::types::primitive::prelude::Address,
395        target: SessionTarget,
396        cfg: hopr_transport::SessionClientConfig,
397    ) -> Result<(HoprSession, HoprSessionConfigurator), HoprLibError> {
398        self.error_if_not_in_state(HoprState::Running, "Node is not ready for on-chain operations".into())?;
399
400        let backoff = backon::ConstantBuilder::default()
401            .with_max_times(self.cfg.protocol.session.establish_max_retries)
402            .with_delay(self.cfg.protocol.session.establish_retry_timeout)
403            .with_jitter();
404
405        use backon::Retryable;
406
407        Ok((|| {
408            let cfg = cfg.clone();
409            let target = target.clone();
410            async { self.transport_api.new_session(destination, target, cfg).await }
411        })
412        .retry(backoff)
413        .sleep(backon::FuturesTimerSleeper)
414        .await?)
415    }
416
417    /// Opens a session using explicit intermediate paths for forward and return routing.
418    #[cfg(all(feature = "session-client", feature = "explicit-path"))]
419    #[allow(deprecated)]
420    #[deprecated(
421        since = "4.0.2-rc.1",
422        note = "temporary compatibility API; remove once the explicit path is not needed anymore."
423    )]
424    pub async fn connect_to_using_explicit_path(
425        &self,
426        destination: hopr_api::types::primitive::prelude::Address,
427        target: SessionTarget,
428        cfg: HoprSessionClientExplicitPathConfig,
429    ) -> Result<(HoprSession, HoprSessionConfigurator), HoprLibError> {
430        let transport_cfg = hopr_transport::SessionClientConfig::try_from(cfg)
431            .map_err(|error| HoprLibError::GeneralError(error.to_string()))?;
432        self.connect_to_with_transport_config(destination, target, transport_cfg)
433            .await
434    }
435}
436
437#[cfg(feature = "session-client")]
438#[async_trait::async_trait]
439impl<Chain, Graph, Net, TMgr> hopr_api::node::HoprSessionClientOperations for Hopr<Chain, Graph, Net, TMgr>
440where
441    Chain: HoprChainApi + Clone + Send + Sync + 'static,
442    Graph: HoprGraphApi<HoprNodeId = OffchainPublicKey> + Clone + Send + Sync + 'static,
443    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed:
444        hopr_api::graph::traits::EdgeObservableRead + Send + 'static,
445    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
446    Net: hopr_api::network::NetworkView + NetworkStreamControl + Send + Sync + Clone + 'static,
447    TMgr: Send + Sync + 'static,
448{
449    type Config = HoprSessionClientConfig;
450    type Error = HoprLibError;
451    type Session = HoprSession;
452    type SessionConfigurator = HoprSessionConfigurator;
453    type Target = SessionTarget;
454
455    async fn connect_to(
456        &self,
457        destination: hopr_api::types::primitive::prelude::Address,
458        target: Self::Target,
459        cfg: Self::Config,
460    ) -> Result<(Self::Session, Self::SessionConfigurator), Self::Error> {
461        self.connect_to_with_transport_config(destination, target, hopr_transport::SessionClientConfig::from(cfg))
462            .await
463    }
464}
465
466// ---------------------------------------------------------------------------
467// Has* accessor trait implementations
468// ---------------------------------------------------------------------------
469
470/// Maps [`Health`] into a [`ComponentStatus`] for a named component.
471fn network_health_to_status(health: Health, component: &str) -> ComponentStatus {
472    match health {
473        Health::Green | Health::Yellow => ComponentStatus::Ready,
474        Health::Orange => ComponentStatus::Degraded(format!("{component}: low connectivity (1 peer)").into()),
475        // Red is returned both for "zero peers" and "network not initialized"
476        Health::Red | Health::Unknown => {
477            ComponentStatus::Unavailable(format!("{component}: no connected peers").into())
478        }
479    }
480}
481
482impl<Chain, Graph, Net, TMgr> HasChainApi for Hopr<Chain, Graph, Net, TMgr>
483where
484    Chain: HoprChainApi + ComponentStatusReporter + Clone + Send + Sync + 'static,
485{
486    type ChainApi = Chain;
487    type ChainError = HoprLibError;
488
489    fn identity(&self) -> &NodeOnchainIdentity {
490        &self.chain_id
491    }
492
493    fn chain_api(&self) -> &Chain {
494        &self.chain_api
495    }
496
497    fn status(&self) -> ComponentStatus {
498        self.chain_api.component_status()
499    }
500
501    fn wait_for_on_chain_event<F>(
502        &self,
503        predicate: F,
504        context: String,
505        timeout: Duration,
506    ) -> EventWaitResult<<Self::ChainApi as HoprChainApi>::ChainError, Self::ChainError>
507    where
508        F: Fn(&ChainEvent) -> bool + Send + Sync + 'static,
509    {
510        debug!(%context, "registering wait for on-chain event");
511
512        // DropAbortable not needed because the stream only generates items when polled
513        let (event_stream, handle) = futures::stream::abortable(
514            self.chain_api
515                .subscribe()?
516                .skip_while(move |event| futures::future::ready(!predicate(event))),
517        );
518
519        let ctx = context.clone();
520
521        Ok((
522            spawn(async move {
523                pin_mut!(event_stream);
524                let res = event_stream
525                    .next()
526                    .timeout(futures_time::time::Duration::from(timeout))
527                    .map_err(|_| {
528                        HoprLibError::Timeout {
529                            context: format!("{ctx} (after {timeout:?})"),
530                        }
531                        .into_right()
532                    })
533                    .await?
534                    .ok_or(
535                        HoprLibError::GeneralError(format!("on-chain event stream for {ctx} ended unexpectedly"))
536                            .into_right(),
537                    );
538                debug!(%ctx, ?res, "on-chain event waiting done");
539                res
540            })
541            .map_err(move |_| {
542                HoprLibError::GeneralError(format!("failed to spawn on-chain event wait for {context}")).into_right()
543            })
544            .and_then(futures::future::ready)
545            .boxed(),
546            handle,
547        ))
548    }
549}
550
551impl<Chain, Graph, Net, TMgr> HasNetworkView for Hopr<Chain, Graph, Net, TMgr>
552where
553    Chain: Send + Sync + 'static,
554    Graph: Send + Sync + 'static,
555    Net: hopr_api::network::NetworkView + Send + Sync + 'static,
556{
557    type NetworkView = HoprTransport<Chain, Graph, Net>;
558
559    fn network_view(&self) -> &Self::NetworkView {
560        &self.transport_api
561    }
562
563    fn status(&self) -> ComponentStatus {
564        network_health_to_status(self.transport_api.health(), "network")
565    }
566}
567
568impl<Chain, Graph, Net, TMgr> HasGraphView for Hopr<Chain, Graph, Net, TMgr>
569where
570    Chain: HoprChainApi + Clone + Send + Sync + 'static,
571    Graph: HoprGraphApi<HoprNodeId = OffchainPublicKey>
572        + hopr_api::graph::NetworkGraphConnectivity<NodeId = OffchainPublicKey>
573        + Clone
574        + Send
575        + Sync
576        + 'static,
577    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed:
578        hopr_api::graph::traits::EdgeObservableRead + Send + 'static,
579    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
580    Net: hopr_api::network::NetworkView + NetworkStreamControl + Send + Sync + Clone + 'static,
581{
582    type Graph = Graph;
583
584    fn graph(&self) -> &Graph {
585        self.transport_api.graph()
586    }
587
588    fn status(&self) -> ComponentStatus {
589        ComponentStatus::Ready
590    }
591}
592
593impl<Chain, Graph, Net, TMgr> HasTransportApi for Hopr<Chain, Graph, Net, TMgr>
594where
595    Chain: HoprChainApi + Clone + Send + Sync + 'static,
596    Graph: HoprGraphApi<HoprNodeId = OffchainPublicKey> + Clone + Send + Sync + 'static,
597    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed:
598        hopr_api::graph::traits::EdgeObservableRead + Send + 'static,
599    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
600    Net: hopr_api::network::NetworkView + NetworkStreamControl + Send + Sync + Clone + 'static,
601    TMgr: Send + Sync + 'static,
602{
603    type Transport = HoprTransport<Chain, Graph, Net>;
604
605    fn transport(&self) -> &Self::Transport {
606        &self.transport_api
607    }
608
609    fn status(&self) -> ComponentStatus {
610        network_health_to_status(self.transport_api.health(), "transport")
611    }
612}
613
614// Available only on Relay nodes that specify `TMgr` that implements TicketManagement
615impl<Chain, Graph, Net, TMgr> HasTicketManagement for Hopr<Chain, Graph, Net, TMgr>
616where
617    Chain: HoprChainApi + Clone + Send + Sync + 'static,
618    TMgr: TicketManagement + Clone + Send + Sync + 'static,
619{
620    type TicketManager = TMgr;
621
622    fn ticket_management(&self) -> &TMgr {
623        &self.ticket_manager
624    }
625
626    fn subscribe_ticket_events(&self) -> impl Stream<Item = hopr_api::node::TicketEvent> + Send + 'static {
627        self.ticket_event_subscribers.1.activate_cloned()
628    }
629
630    fn status(&self) -> ComponentStatus {
631        ComponentStatus::Ready
632    }
633}
634
635impl<Chain, Graph, Net, TMgr> hopr_api::node::ActionableEventSource for Hopr<Chain, Graph, Net, TMgr>
636where
637    Chain: HoprChainApi + Send + Sync + 'static,
638    Graph: Send + Sync + 'static,
639    Net: hopr_api::network::NetworkView + Send + Sync + 'static,
640    TMgr: Send + Sync + 'static,
641{
642    fn subscribe_to_actionable_events(
643        &self,
644        filter: Option<&[ActionableEventDiscriminant]>,
645    ) -> Result<futures::stream::BoxStream<'static, ActionableEvent>, String> {
646        let wants = |d: ActionableEventDiscriminant| filter.is_none_or(|f| f.contains(&d));
647
648        let mut streams = Vec::<futures::stream::BoxStream<'static, ActionableEvent>>::new();
649
650        if wants(ActionableEventDiscriminant::Chain) {
651            streams.push(
652                self.chain_api
653                    .subscribe()
654                    .map_err(|e| e.to_string())?
655                    .map(ActionableEvent::Chain)
656                    .boxed(),
657            );
658        }
659
660        if wants(ActionableEventDiscriminant::Network) {
661            streams.push(
662                self.transport_api
663                    .subscribe_network_events()
664                    .map(ActionableEvent::Network)
665                    .boxed(),
666            );
667        }
668
669        if wants(ActionableEventDiscriminant::Ticket) {
670            streams.push(
671                self.ticket_event_subscribers
672                    .1
673                    .activate_cloned()
674                    .map(ActionableEvent::Ticket)
675                    .boxed(),
676            );
677        }
678
679        if streams.is_empty() {
680            return Ok(futures::stream::empty().boxed());
681        }
682
683        // `Merge` provides fair polling distribution across active sources.
684        Ok(streams.merge().boxed())
685    }
686}
687
688/// Per-component status report for the HOPR node.
689#[derive(Debug, Clone)]
690pub struct NodeComponentStatuses {
691    /// Overall node lifecycle state.
692    pub node_state: HoprState,
693    /// Chain/blokli connector status.
694    pub chain: ComponentStatus,
695    /// P2P network layer status.
696    pub network: ComponentStatus,
697    /// Transport layer status.
698    pub transport: ComponentStatus,
699}
700
701impl NodeComponentStatuses {
702    /// Worst-case aggregation: the overall status is the worst of any component.
703    pub fn aggregate(&self) -> ComponentStatus {
704        let statuses = [&self.chain, &self.network, &self.transport];
705        if statuses.iter().any(|s| s.is_unavailable()) {
706            ComponentStatus::Unavailable("one or more components unavailable".into())
707        } else if statuses.iter().any(|s| s.is_degraded()) {
708            ComponentStatus::Degraded("one or more components degraded".into())
709        } else if statuses.iter().any(|s| s.is_initializing()) {
710            ComponentStatus::Initializing("one or more components initializing".into())
711        } else {
712            ComponentStatus::Ready
713        }
714    }
715}
716
717impl<Chain, Graph, Net, TMgr> Hopr<Chain, Graph, Net, TMgr>
718where
719    Chain: HoprChainApi + ComponentStatusReporter + Clone + Send + Sync + 'static,
720    Net: hopr_api::network::NetworkView + NetworkStreamControl + Send + Sync + Clone + 'static,
721    Graph: HoprGraphApi<HoprNodeId = OffchainPublicKey>
722        + hopr_api::graph::NetworkGraphConnectivity<NodeId = OffchainPublicKey>
723        + Clone
724        + Send
725        + Sync
726        + 'static,
727    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed:
728        hopr_api::graph::traits::EdgeObservableRead + Send + 'static,
729    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
730    TMgr: Send + Sync + 'static,
731{
732    /// Returns per-component health statuses for the node.
733    ///
734    /// When the node has reached `Running`, the aggregate `node_state` is
735    /// derived from component statuses (Running → Degraded → Failed).
736    pub fn component_statuses(&self) -> NodeComponentStatuses {
737        let base = self.state.load(Ordering::Relaxed);
738        let statuses = NodeComponentStatuses {
739            node_state: base,
740            chain: HasChainApi::status(self),
741            network: HasNetworkView::status(self),
742            transport: HasTransportApi::status(self),
743        };
744
745        // Derive aggregate HoprState from component statuses once Running
746        if base == HoprState::Running {
747            NodeComponentStatuses {
748                node_state: match statuses.aggregate() {
749                    ComponentStatus::Unavailable(_) => HoprState::Failed,
750                    ComponentStatus::Degraded(_) | ComponentStatus::Initializing(_) => HoprState::Degraded,
751                    ComponentStatus::Ready => HoprState::Running,
752                },
753                ..statuses
754            }
755        } else {
756            statuses
757        }
758    }
759}
760
761impl<Chain, Graph, Net, TMgr> HoprNodeOperations for Hopr<Chain, Graph, Net, TMgr> {
762    fn status(&self) -> HoprState {
763        self.state.load(Ordering::Relaxed)
764    }
765}
766
767impl<Chain, Graph, Net, TMgr> hopr_api::node::PacketTransport for Hopr<Chain, Graph, Net, TMgr> {
768    fn packet_payload_size() -> usize {
769        hopr_transport::PACKET_PAYLOAD_SIZE
770    }
771}
772
773/// Prometheus-formatted metrics collected by the hopr-lib components.
774///
775/// Only available when compiled with the `telemetry` feature.
776#[cfg(feature = "telemetry")]
777pub fn collect_hopr_metrics() -> errors::Result<String> {
778    hopr_api::types::telemetry::gather_all_metrics().map_err(HoprLibError::other)
779}
780
781/// Converts a PeerId to an OffchainPublicKey.
782///
783/// This is a standalone utility function, not part of the API traits.
784pub fn peer_id_to_offchain_key(peer_id: &PeerId) -> errors::Result<OffchainPublicKey> {
785    Ok(hopr_transport::peer_id_to_public_key(peer_id)?)
786}
787
788#[cfg(test)]
789mod tests {
790    use super::*;
791
792    #[test]
793    fn network_health_green_is_ready() {
794        assert_eq!(network_health_to_status(Health::Green, "test"), ComponentStatus::Ready);
795    }
796
797    #[test]
798    fn network_health_yellow_is_ready() {
799        assert_eq!(network_health_to_status(Health::Yellow, "test"), ComponentStatus::Ready);
800    }
801
802    #[test]
803    fn network_health_orange_is_degraded() {
804        assert!(network_health_to_status(Health::Orange, "network").is_degraded());
805    }
806
807    #[test]
808    fn network_health_red_is_unavailable() {
809        assert!(network_health_to_status(Health::Red, "network").is_unavailable());
810    }
811
812    #[test]
813    fn network_health_unknown_is_unavailable() {
814        assert!(network_health_to_status(Health::Unknown, "network").is_unavailable());
815    }
816
817    #[test]
818    fn aggregate_all_ready() {
819        let statuses = NodeComponentStatuses {
820            node_state: HoprState::Running,
821            chain: ComponentStatus::Ready,
822            network: ComponentStatus::Ready,
823            transport: ComponentStatus::Ready,
824        };
825        assert_eq!(statuses.aggregate(), ComponentStatus::Ready);
826    }
827
828    #[test]
829    fn aggregate_one_degraded() {
830        let statuses = NodeComponentStatuses {
831            node_state: HoprState::Running,
832            chain: ComponentStatus::Ready,
833            network: ComponentStatus::Degraded("low peers".into()),
834            transport: ComponentStatus::Ready,
835        };
836        assert!(statuses.aggregate().is_degraded());
837    }
838
839    #[test]
840    fn aggregate_one_unavailable() {
841        let statuses = NodeComponentStatuses {
842            node_state: HoprState::Running,
843            chain: ComponentStatus::Unavailable("blokli down".into()),
844            network: ComponentStatus::Ready,
845            transport: ComponentStatus::Ready,
846        };
847        assert!(statuses.aggregate().is_unavailable());
848    }
849
850    #[test]
851    fn aggregate_unavailable_wins_over_degraded() {
852        let statuses = NodeComponentStatuses {
853            node_state: HoprState::Running,
854            chain: ComponentStatus::Unavailable("blokli down".into()),
855            network: ComponentStatus::Degraded("low peers".into()),
856            transport: ComponentStatus::Ready,
857        };
858        assert!(statuses.aggregate().is_unavailable());
859    }
860
861    #[test]
862    fn aggregate_one_initializing() {
863        let statuses = NodeComponentStatuses {
864            node_state: HoprState::Running,
865            chain: ComponentStatus::Initializing("starting".into()),
866            network: ComponentStatus::Ready,
867            transport: ComponentStatus::Ready,
868        };
869        assert!(statuses.aggregate().is_initializing());
870    }
871
872    #[test]
873    fn aggregate_degraded_wins_over_initializing() {
874        let statuses = NodeComponentStatuses {
875            node_state: HoprState::Running,
876            chain: ComponentStatus::Initializing("starting".into()),
877            network: ComponentStatus::Degraded("low peers".into()),
878            transport: ComponentStatus::Ready,
879        };
880        assert!(statuses.aggregate().is_degraded());
881    }
882
883    #[test]
884    fn network_health_to_status_includes_component_name() {
885        match network_health_to_status(Health::Orange, "mycomp") {
886            ComponentStatus::Degraded(d) => assert!(d.contains("mycomp"), "detail should contain component name"),
887            other => panic!("expected Degraded, got {other:?}"),
888        }
889    }
890
891    #[test]
892    fn network_health_to_status_red_and_unknown_are_same_variant() {
893        let red = network_health_to_status(Health::Red, "x");
894        let unknown = network_health_to_status(Health::Unknown, "x");
895        assert!(red.is_unavailable());
896        assert!(unknown.is_unavailable());
897    }
898
899    #[cfg(all(feature = "session-client", feature = "explicit-path"))]
900    #[allow(deprecated)]
901    #[test]
902    fn explicit_path_config_converts_into_intermediate_path_routing_options() -> anyhow::Result<()> {
903        use anyhow::Context as _;
904        use hopr_transport::Keypair;
905
906        let k1 = hopr_api::types::internal::NodeId::from(*OffchainKeypair::random().public());
907        let k2 = hopr_api::types::internal::NodeId::from(*OffchainKeypair::random().public());
908        let k3 = hopr_api::types::internal::NodeId::from(*OffchainKeypair::random().public());
909
910        let cfg = hopr_transport::SessionClientConfig::try_from(HoprSessionClientExplicitPathConfig {
911            forward_path: vec![k1, k2],
912            return_path: vec![k3],
913            capabilities: SessionCapability::Segmentation.into(),
914            pseudonym: None,
915            surb_management: None,
916            always_max_out_surbs: false,
917            flow_control: None,
918            max_frames_behind_gap: Some(8),
919        })
920        .context("explicit path config conversion must succeed")?;
921
922        assert_eq!(
923            cfg.max_frames_behind_gap,
924            Some(8),
925            "the session's head-of-line bound has to survive the conversion, or it silently reverts to the node \
926             default"
927        );
928        assert!(matches!(
929            cfg.forward_path_options,
930            hopr_transport::RoutingOptions::IntermediatePath(_)
931        ));
932        assert!(matches!(
933            cfg.return_path_options,
934            hopr_transport::RoutingOptions::IntermediatePath(_)
935        ));
936        Ok(())
937    }
938}
hopr_lib/lib.rs

hopr_lib/
lib.rs
