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

hopr_lib/
lib.rs
