hopr_lib/
builder.rs

1//! Type-state builder for constructing a [`Hopr`] node.
2//!
3//! The builder guides construction through a series of mandatory phases:
4//!
5//! 1. **Identity** — `HoprBuilder` → `HoprBuilder::with_identity`
6//! 2. **Configuration** — `HoprBuilderWithIdentity::with_config`
7//! 3. **Component factories** — chain API, graph, network, and cover-traffic
8//! 4. **Session server** (when the `session-server` feature is enabled) — `HoprBuilderConfigured::with_session_server`
9//! 5. **Build** — `build_edge` for an entry/exit node or `build_full` for a relay node with ticket management.
10//!
11//! # Example
12//!
13//! ```rust,ignore
14//! use hopr_lib::{config::HoprLibConfig, builder::{HoprBuilder, ChainKeypair, OffchainKeypair, Keypair}};
15//!
16//! let chain_key = ChainKeypair::random();
17//! let offchain_key = OffchainKeypair::random();
18//! let config = HoprLibConfig::default();
19//!
20//! let builder = HoprBuilder
21//!     .with_identity(&chain_key, &offchain_key)
22//!     .with_config(config)
23//!     .with_chain_api(|_ctx| { /* ... */ todo!() })
24//!     .with_graph(|_ctx| { /* ... */ todo!() })
25//!     .with_network(|_ctx| Box::pin(async { /* ... */ Ok(todo!()) }))
26//!     .with_cover_traffic(|_ctx| { /* ... */ todo!() });
27//! ```
28
29mod chain_wiring;
30
31use std::{future::Future, pin::Pin, sync::Arc, time::Duration};
32
33use futures::{FutureExt, StreamExt};
34pub use hopr_api::types::crypto::{
35    keypairs::Keypair,
36    prelude::{ChainKeypair, OffchainKeypair},
37};
38use hopr_api::{
39    chain::{AnnouncementError, HoprChainApi, SafeRegistrationError, StateSyncOptions},
40    ct::{CoverTrafficGeneration, ProbingTrafficGeneration},
41    graph::HoprGraphApi,
42    network::{BoxedProcessFn, NetworkStreamControl, NetworkView},
43    node::{AtomicHoprState, HoprState, NodeOnchainIdentity, PixEvent, TicketEvent},
44    tickets::{TicketFactory, TicketManagement},
45    types::{chain::chain_events::ChainEvent, internal::prelude::ChannelDirection, primitive::prelude::Address},
46};
47use hopr_transport::{HoprTransport, IncomingSession};
48use hopr_utils::{
49    network_types::{
50        addr::is_public_address,
51        crossfire_sink::{CrossfireSink, bounded_sink_channel},
52    },
53    runtime::{AbortableList, prelude::spawn},
54};
55use validator::Validate;
56
57use crate::{
58    Hopr, HoprLibError, HoprLibProcess, MIN_NATIVE_BALANCE, NODE_READY_TIMEOUT, SUGGESTED_NATIVE_BALANCE,
59    config::HoprLibConfig, constants, helpers::BroadcastSenderSink,
60};
61
62#[cfg(all(feature = "telemetry", not(test)))]
63lazy_static::lazy_static! {
64    static ref METRIC_PROCESS_START_TIME:  hopr_api::types::telemetry::SimpleGauge =  hopr_api::types::telemetry::SimpleGauge::new(
65        "hopr_start_time",
66        "The unix timestamp in seconds at which the process was started"
67    ).unwrap();
68    static ref METRIC_HOPR_LIB_VERSION:  hopr_api::types::telemetry::MultiGauge =  hopr_api::types::telemetry::MultiGauge::new(
69        "hopr_lib_version",
70        "Executed version of hopr-lib",
71        &["version"]
72    ).unwrap();
73    static ref METRIC_HOPR_NODE_INFO:  hopr_api::types::telemetry::MultiGauge =  hopr_api::types::telemetry::MultiGauge::new(
74        "hopr_node_addresses",
75        "Node on-chain and off-chain addresses",
76        &["peerid", "address", "safe_address", "module_address"]
77    ).unwrap();
78}
79
80const PEER_DISCOVERY_CHANNEL_CAPACITY: usize = 2048;
81
82type PeerDiscoveryRx =
83    Arc<parking_lot::Mutex<Option<futures::stream::BoxStream<'static, (hopr_api::PeerId, Vec<hopr_api::Multiaddr>)>>>>;
84
85/// What `into_parts` hands to the shared build path: the configured builder, the two channels the
86/// transport is wired to, and the tasks already spawned for them. The admission sink is optional
87/// because a node built without a session server has nobody to ask.
88type BuildParts<Chain, Graph, Net, Ct> = (
89    HoprBuilderConfigured<Chain, Graph, Net, Ct>,
90    futures::channel::mpsc::Sender<IncomingSession>,
91    Option<hopr_transport::SessionAdmissionSink>,
92    AbortableList<HoprLibProcess>,
93);
94
95/// Type-erased factory closure producing `T` from a [`BuildCtx`] reference.
96type Factory<T> = Box<dyn FnOnce(&BuildCtx) -> T + Send>;
97type AsyncFactory<T> = Box<dyn FnOnce(BuildCtx) -> Pin<Box<dyn Future<Output = T> + Send>> + Send>;
98
99/// Context available to factory closures during the build step.
100#[derive(Clone)]
101pub struct BuildCtx {
102    /// Node's on-chain keypair.
103    pub chain_key: ChainKeypair,
104    /// Node's off-chain (packet) keypair.
105    pub packet_key: OffchainKeypair,
106    /// Node configuration.
107    pub cfg: HoprLibConfig,
108    peer_discovery_rx: PeerDiscoveryRx,
109}
110
111impl BuildCtx {
112    /// Take the peer-discovery receiver. Returns `Some` on the first call, `None` afterwards.
113    pub fn take_peer_discovery_rx(
114        &self,
115    ) -> Option<futures::stream::BoxStream<'static, (hopr_api::PeerId, Vec<hopr_api::Multiaddr>)>> {
116        self.peer_discovery_rx.lock().take()
117    }
118}
119
120// ---------------------------------------------------------------------------
121// Type-state builder phases
122// ---------------------------------------------------------------------------
123
124/// Initial builder — forces `with_identity` first.
125#[derive(Default)]
126pub struct HoprBuilder;
127
128impl HoprBuilder {
129    /// Sets the node's on-chain and off-chain identity.
130    pub fn with_identity(self, chain_key: &ChainKeypair, offchain_key: &OffchainKeypair) -> HoprBuilderWithIdentity {
131        HoprBuilderWithIdentity {
132            chain_key: chain_key.clone(),
133            packet_key: offchain_key.clone(),
134        }
135    }
136}
137
138/// Builder with identity set — forces `with_config` next.
139pub struct HoprBuilderWithIdentity {
140    chain_key: ChainKeypair,
141    packet_key: OffchainKeypair,
142}
143
144impl HoprBuilderWithIdentity {
145    /// Sets the node configuration and produces the configured builder.
146    pub fn with_config(self, cfg: HoprLibConfig) -> HoprBuilderConfigured {
147        let (peer_discovery_tx, peer_discovery_rx) =
148            bounded_sink_channel::<(hopr_api::PeerId, Vec<hopr_api::Multiaddr>)>(PEER_DISCOVERY_CHANNEL_CAPACITY);
149        HoprBuilderConfigured {
150            ctx: BuildCtx {
151                chain_key: self.chain_key,
152                packet_key: self.packet_key,
153                cfg,
154                peer_discovery_rx: Arc::new(parking_lot::Mutex::new(Some(peer_discovery_rx))),
155            },
156            safe_and_module: None,
157            chain_factory: None,
158            graph_factory: None,
159            network_factory: None,
160            ct_factory: None,
161            peer_discovery_tx,
162        }
163    }
164}
165
166// ---------------------------------------------------------------------------
167// HoprBuilderConfigured — stores factories, no session yet
168// ---------------------------------------------------------------------------
169
170/// Configured builder accepting factory closures for components.
171///
172/// When the `session-server` feature is enabled, `with_session_server`
173/// must be called before building — it returns a `HoprBuilderWithSession` which
174/// has the `build_edge` / `build_full` methods.
175///
176/// When the feature is disabled, `build_edge` / `build_full` are available directly.
177pub struct HoprBuilderConfigured<Chain = (), Graph = (), Net = (), Ct = ()> {
178    ctx: BuildCtx,
179    safe_and_module: Option<(Address, Address)>,
180    chain_factory: Option<Factory<Chain>>,
181    graph_factory: Option<Factory<Graph>>,
182    network_factory: Option<AsyncFactory<Result<(Net, BoxedProcessFn), HoprLibError>>>,
183    ct_factory: Option<Factory<Ct>>,
184    peer_discovery_tx: CrossfireSink<(hopr_api::PeerId, Vec<hopr_api::Multiaddr>)>,
185}
186
187impl<Chain, Graph, Net, Ct> HoprBuilderConfigured<Chain, Graph, Net, Ct> {
188    /// Sets the node Safe and module addresses.
189    pub fn with_safe_module(mut self, safe: &Address, module: &Address) -> Self {
190        self.safe_and_module = Some((*safe, *module));
191        self
192    }
193
194    /// Sets the chain API factory.
195    pub fn with_chain_api<NewChain>(
196        self,
197        f: impl FnOnce(&BuildCtx) -> NewChain + Send + 'static,
198    ) -> HoprBuilderConfigured<NewChain, Graph, Net, Ct> {
199        HoprBuilderConfigured {
200            ctx: self.ctx,
201            safe_and_module: self.safe_and_module,
202            chain_factory: Some(Box::new(f)),
203            graph_factory: self.graph_factory,
204            network_factory: self.network_factory,
205            ct_factory: self.ct_factory,
206            peer_discovery_tx: self.peer_discovery_tx,
207        }
208    }
209
210    /// Sets the graph factory.
211    pub fn with_graph<NewGraph>(
212        self,
213        f: impl FnOnce(&BuildCtx) -> NewGraph + Send + 'static,
214    ) -> HoprBuilderConfigured<Chain, NewGraph, Net, Ct> {
215        HoprBuilderConfigured {
216            ctx: self.ctx,
217            safe_and_module: self.safe_and_module,
218            chain_factory: self.chain_factory,
219            graph_factory: Some(Box::new(f)),
220            network_factory: self.network_factory,
221            ct_factory: self.ct_factory,
222            peer_discovery_tx: self.peer_discovery_tx,
223        }
224    }
225
226    /// Sets the network factory. Must resolve to `Ok((Net, BoxedProcessFn))`.
227    ///
228    /// The factory receives [`BuildCtx`] by value and returns a boxed future,
229    /// allowing async network construction without blocking the executor.
230    /// Failures are propagated as [`HoprLibError`] during the build step.
231    pub fn with_network<NewNet>(
232        self,
233        f: impl FnOnce(BuildCtx) -> Pin<Box<dyn Future<Output = Result<(NewNet, BoxedProcessFn), HoprLibError>> + Send>>
234        + Send
235        + 'static,
236    ) -> HoprBuilderConfigured<Chain, Graph, NewNet, Ct> {
237        HoprBuilderConfigured {
238            ctx: self.ctx,
239            safe_and_module: self.safe_and_module,
240            chain_factory: self.chain_factory,
241            graph_factory: self.graph_factory,
242            network_factory: Some(Box::new(f)),
243            ct_factory: self.ct_factory,
244            peer_discovery_tx: self.peer_discovery_tx,
245        }
246    }
247
248    /// Sets the cover traffic factory.
249    pub fn with_cover_traffic<NewCt>(
250        self,
251        f: impl FnOnce(&BuildCtx) -> NewCt + Send + 'static,
252    ) -> HoprBuilderConfigured<Chain, Graph, Net, NewCt> {
253        HoprBuilderConfigured {
254            ctx: self.ctx,
255            safe_and_module: self.safe_and_module,
256            chain_factory: self.chain_factory,
257            graph_factory: self.graph_factory,
258            network_factory: self.network_factory,
259            ct_factory: Some(Box::new(f)),
260            peer_discovery_tx: self.peer_discovery_tx,
261        }
262    }
263
264    /// Attaches a session server for handling incoming sessions.
265    ///
266    /// Eagerly spawns the server task and returns a [`HoprBuilderWithSession`]
267    /// that has the `build_edge` / `build_full` methods.
268    ///
269    /// Two tasks are spawned, not one, because the server is consulted at two different points in a
270    /// Session's life. `admit` is asked while the peer's request is still being negotiated and no
271    /// Session exists yet; `process` is handed the established byte-stream. Sharing one channel
272    /// would put a long-lived `process` in front of the admission decisions queued behind it.
273    #[cfg(feature = "session-server")]
274    pub fn with_session_server(
275        self,
276        // `Sync` on top of what `process` alone needed: the server is now driven from two
277        // independent tasks, and `admit` carries a default body, which `async_trait` can only make
278        // `Send` by requiring `&Self` to cross threads.
279        server: impl hopr_api::node::HoprSessionServer<Session = IncomingSession, Error: std::fmt::Display>
280        + Clone
281        + Send
282        + Sync
283        + 'static,
284    ) -> HoprBuilderWithSession<Chain, Graph, Net, Ct> {
285        let incoming_session_capacity = self.ctx.cfg.incoming_session_capacity.max(1);
286
287        let (session_tx, session_rx) = futures::channel::mpsc::channel::<IncomingSession>(incoming_session_capacity);
288
289        // Sized like the incoming-session channel: an admission request is in flight only for as
290        // long as one Start negotiation, so this bounds concurrent negotiations rather than live
291        // Sessions. Overflowing refuses the Session as busy rather than queueing its peer.
292        let (admission_tx, admission_rx) = futures::channel::mpsc::channel(incoming_session_capacity);
293
294        let admission_server = server.clone();
295        let admission_diag = hopr_utils::runtime::diagnostics::ConcurrentDiagnostics::new(
296            "session_admission_for_each_concurrent",
297            module_path!(),
298            file!(),
299            line!(),
300        );
301        let admission_handle = hopr_utils::spawn_as_abortable_named!(
302            "hopr_lib_session_admission",
303            admission_rx
304                .for_each_concurrent(
305                    None,
306                    move |(request, reply): (_, hopr_transport::SessionAdmissionReply)| {
307                        let server = admission_server.clone();
308                        let admission_diag = admission_diag.clone();
309                        admission_diag.wrap(|| async move {
310                            // Concurrently rather than in sequence: the transport gives each request its
311                            // own deadline, so one slow decision must not spend another's.
312                            let decision = server.admit(request).await.map_err(|error| error.to_string());
313                            if reply.send(decision).is_err() {
314                                // The transport stopped waiting — its own timeout fired, or the peer
315                                // went away. Nothing to do but note it; it already refused the Session.
316                                tracing::debug!("session admission decided after the transport stopped waiting");
317                            }
318                        })
319                    }
320                )
321                .inspect(|_| tracing::warn!(
322                    task = %HoprLibProcess::SessionAdmission,
323                    "long-running background task finished"
324                ))
325        );
326
327        tracing::debug!(capacity = incoming_session_capacity, "spawning session server");
328        let session_diag = hopr_utils::runtime::diagnostics::ConcurrentDiagnostics::new(
329            "session_server_for_each_concurrent",
330            module_path!(),
331            file!(),
332            line!(),
333        );
334        let handle = hopr_utils::spawn_as_abortable_named!(
335            "hopr_lib_session_server",
336            session_rx
337                .for_each_concurrent(None, move |session| {
338                    let server = server.clone();
339                    let session_diag = session_diag.clone();
340                    session_diag.wrap(|| async move {
341                        let session_id = *session.session.id();
342                        match server.process(session).await {
343                            Ok(()) => tracing::debug!(?session_id, "session processed successfully"),
344                            Err(error) => {
345                                tracing::error!(?session_id, %error, "session processing failed")
346                            }
347                        }
348                    })
349                })
350                .inspect(|_| tracing::warn!(
351                    task = %HoprLibProcess::SessionServer,
352                    "long-running background task finished"
353                ))
354        );
355
356        HoprBuilderWithSession {
357            inner: self,
358            session_tx,
359            admission_tx,
360            session_handle: handle,
361            admission_handle,
362        }
363    }
364}
365
366// ---------------------------------------------------------------------------
367// HoprBuilderWithSession — session server attached, ready to build
368// ---------------------------------------------------------------------------
369
370/// Builder with a session server attached. Has `build_edge` / `build_full`.
371///
372/// Only exists when the `session-server` feature is enabled.
373#[cfg(feature = "session-server")]
374pub struct HoprBuilderWithSession<Chain = (), Graph = (), Net = (), Ct = ()> {
375    inner: HoprBuilderConfigured<Chain, Graph, Net, Ct>,
376    session_tx: futures::channel::mpsc::Sender<IncomingSession>,
377    admission_tx: hopr_transport::SessionAdmissionSink,
378    session_handle: hopr_utils::runtime::AbortHandle,
379    admission_handle: hopr_utils::runtime::AbortHandle,
380}
381
382// ---------------------------------------------------------------------------
383// Intermediate pre-build state
384// ---------------------------------------------------------------------------
385
386struct PreHopr<Chain, Graph, Net, Ct> {
387    chain_id: ChainKeypair,
388    transport_id: OffchainKeypair,
389    cfg: HoprLibConfig,
390    state: Arc<AtomicHoprState>,
391    transport_api: HoprTransport<Chain, Graph, Net>,
392    chain_api: Chain,
393
394    ticket_event_subscribers: (
395        async_broadcast::Sender<TicketEvent>,
396        async_broadcast::InactiveReceiver<TicketEvent>,
397    ),
398    pix_event_subscribers: (
399        async_broadcast::Sender<PixEvent>,
400        async_broadcast::InactiveReceiver<PixEvent>,
401    ),
402    processes: AbortableList<HoprLibProcess>,
403    session_tx: futures::channel::mpsc::Sender<IncomingSession>,
404    admission_tx: Option<hopr_transport::SessionAdmissionSink>,
405    cover_traffic: Ct,
406    network: Net,
407    network_process: BoxedProcessFn,
408}
409
410// ---------------------------------------------------------------------------
411// Shared pre_build logic
412// ---------------------------------------------------------------------------
413
414/// Drains a HoprSocket reader, discarding all packets and logging throughput every ~60 seconds.
415/// Runs until the stream ends (sender side dropped).
416async fn drain_incoming_data<S: futures::Stream + Unpin>(mut reader: S) {
417    let mut received: u64 = 0;
418    let mut last_report = std::time::Instant::now();
419    while reader.next().await.is_some() {
420        received += 1;
421        if last_report.elapsed().as_secs() >= 60 {
422            tracing::info!(
423                received,
424                "incoming-data drain: unrelated packets discarded in last ~1 min"
425            );
426            received = 0;
427            last_report = std::time::Instant::now();
428        }
429    }
430}
431
432async fn pre_build_inner<Chain, Graph, Net, Ct>(
433    configured: HoprBuilderConfigured<Chain, Graph, Net, Ct>,
434    session_tx: futures::channel::mpsc::Sender<IncomingSession>,
435    admission_tx: Option<hopr_transport::SessionAdmissionSink>,
436    mut processes: AbortableList<HoprLibProcess>,
437) -> Result<PreHopr<Chain, Graph, Net, Ct>, HoprLibError>
438where
439    Chain: HoprChainApi + Clone + Send + Sync + 'static,
440    Graph: HoprGraphApi<HoprNodeId = hopr_api::OffchainPublicKey> + Clone + Send + Sync + 'static,
441    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed:
442        hopr_api::graph::traits::EdgeObservableRead + Send + 'static,
443    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
444    Net: NetworkView + NetworkStreamControl + Send + Sync + Clone + 'static,
445    Ct: ProbingTrafficGeneration + CoverTrafficGeneration + Send + Sync + 'static,
446{
447    let peer_discovery_tx = Some(configured.peer_discovery_tx);
448    let ctx = configured.ctx;
449    ctx.cfg.validate()?;
450
451    let chain_api = (configured
452        .chain_factory
453        .ok_or(HoprLibError::BuilderError("missing chain factory"))?)(&ctx);
454    let graph = (configured
455        .graph_factory
456        .ok_or(HoprLibError::BuilderError("missing graph factory"))?)(&ctx);
457    let (network, network_process) =
458        (configured
459            .network_factory
460            .ok_or(HoprLibError::BuilderError("missing network factory"))?)(ctx.clone())
461        .await?;
462    let cover_traffic = (configured
463        .ct_factory
464        .ok_or(HoprLibError::BuilderError("missing cover traffic factory"))?)(&ctx);
465
466    let (chain_id, transport_id) = (ctx.chain_key.clone(), ctx.packet_key.clone());
467
468    let transport_api = HoprTransport::new(
469        (&chain_id, &transport_id),
470        chain_api.clone(),
471        graph.clone(),
472        vec![(&ctx.cfg.host).try_into().map_err(HoprLibError::TransportError)?],
473        ctx.cfg.protocol.clone(),
474    )
475    .map_err(HoprLibError::TransportError)?;
476
477    #[cfg(all(feature = "telemetry", not(test)))]
478    {
479        use hopr_api::types::primitive::traits::AsUnixTimestamp;
480        METRIC_PROCESS_START_TIME.set(std::time::SystemTime::now().as_unix_timestamp().as_secs_f64());
481        METRIC_HOPR_LIB_VERSION.set(
482            &[const_format::formatcp!("{}", constants::APP_VERSION)],
483            const_format::formatcp!(
484                "{}.{}",
485                env!("CARGO_PKG_VERSION_MAJOR"),
486                env!("CARGO_PKG_VERSION_MINOR")
487            )
488            .parse()
489            .unwrap_or(0.0),
490        );
491    }
492
493    let (mut new_tickets_tx, new_tickets_rx) = async_broadcast::broadcast(2048);
494    new_tickets_tx.set_await_active(false);
495    new_tickets_tx.set_overflow(true);
496
497    let (mut ssa_tx, ssa_rx) = async_broadcast::broadcast(2048);
498    ssa_tx.set_await_active(false);
499    ssa_tx.set_overflow(true);
500
501    let me_onchain = chain_id.public().to_address();
502
503    #[cfg(feature = "testing")]
504    tracing::warn!("!! FOR TESTING ONLY !! Node is running with some safety checks disabled!");
505
506    tracing::info!(
507        address = %me_onchain,
508        minimum_balance = %*SUGGESTED_NATIVE_BALANCE,
509        "node is not started, please fund this node",
510    );
511
512    tracing::info!(
513        suggested_minimum_balance = %*SUGGESTED_NATIVE_BALANCE,
514        "node about to start, checking for funds",
515    );
516    let funding_timeout = Duration::from_secs(200);
517    crate::helpers::wait_for_balance(*MIN_NATIVE_BALANCE, funding_timeout, me_onchain, &chain_api)
518        .await
519        .map_err(|_| {
520            HoprLibError::InsufficientFunds(format!(
521                "failed to fund the node within {} seconds",
522                funding_timeout.as_secs()
523            ))
524        })?;
525
526    tracing::info!("starting HOPR node...");
527    let balance: hopr_api::types::primitive::prelude::XDaiBalance =
528        chain_api.balance(me_onchain).await.map_err(HoprLibError::chain)?;
529    let minimum_balance = *constants::MIN_NATIVE_BALANCE;
530
531    tracing::info!(address = %me_onchain, %balance, %minimum_balance, "node information");
532
533    if balance.le(&minimum_balance) {
534        return Err(HoprLibError::InsufficientFunds(
535            "cannot start the node without a sufficiently funded wallet".into(),
536        ));
537    }
538
539    #[cfg(debug_assertions)]
540    let skip_protocol_checks = ctx.cfg.disable_protocol_checks;
541    #[cfg(not(debug_assertions))]
542    let skip_protocol_checks = false;
543
544    let network_min_ticket_price = chain_api.minimum_ticket_price().await.map_err(HoprLibError::chain)?;
545    let configured_ticket_price = ctx.cfg.protocol.packet.codec.outgoing_ticket_price;
546    if !skip_protocol_checks && configured_ticket_price.is_some_and(|c| c < network_min_ticket_price) {
547        return Err(HoprLibError::GeneralError(format!(
548            "configured outgoing ticket price < network minimum: {configured_ticket_price:?} < \
549             {network_min_ticket_price}"
550        )));
551    }
552
553    let network_min_win_prob = chain_api
554        .minimum_incoming_ticket_win_prob()
555        .await
556        .map_err(HoprLibError::chain)?;
557    let configured_win_prob = ctx.cfg.protocol.packet.codec.outgoing_win_prob;
558
559    if !skip_protocol_checks && configured_win_prob.is_some_and(|c| c.approx_cmp(&network_min_win_prob).is_lt()) {
560        return Err(HoprLibError::GeneralError(format!(
561            "configured outgoing win probability < network minimum: {configured_win_prob:?} < {network_min_win_prob}"
562        )));
563    }
564
565    tracing::info!(
566        peer_id = %transport_id.public().to_peerid_str(),
567        address = %me_onchain,
568        hopr_lib_version = constants::APP_VERSION,
569        "Node information"
570    );
571
572    let safe_addr = ctx.cfg.safe_module.safe_address;
573    if me_onchain == safe_addr {
574        return Err(HoprLibError::GeneralError(
575            "cannot use self as staking safe address".into(),
576        ));
577    }
578
579    tracing::info!(%safe_addr, "registering safe with this node");
580    match chain_api.register_safe(&safe_addr).await {
581        Ok(awaiter) => {
582            awaiter.await.map_err(|error| {
583                tracing::error!(%safe_addr, %error, "safe registration failed");
584                HoprLibError::chain(error)
585            })?;
586            tracing::info!(%safe_addr, "safe successfully registered with this node");
587        }
588        Err(SafeRegistrationError::AlreadyRegistered(registered_safe)) => {
589            if registered_safe == safe_addr {
590                tracing::info!(%safe_addr, "this safe is already registered with this node");
591            } else {
592                tracing::error!(%safe_addr, %registered_safe, "node registered with different safe");
593                return Err(HoprLibError::GeneralError("node registered with different safe".into()));
594            }
595        }
596        Err(error) => {
597            tracing::error!(%safe_addr, %error, "safe registration failed");
598            return Err(HoprLibError::chain(error));
599        }
600    }
601
602    let multiaddresses_to_announce = if ctx.cfg.publish {
603        transport_api.announceable_multiaddresses()
604    } else {
605        Vec::new()
606    };
607
608    multiaddresses_to_announce
609        .iter()
610        .filter(|a| !is_public_address(a))
611        .for_each(|multi_addr| tracing::warn!(?multi_addr, "announcing private multiaddress"));
612
613    // Preflight for the on-chain announcement
614    if ctx.cfg.publish {
615        let key_binding_fee = chain_api.key_binding_fee().await.map_err(HoprLibError::chain)?;
616        if !key_binding_fee.is_zero() {
617            let safe_hopr_balance: hopr_api::types::primitive::prelude::HoprBalance =
618                chain_api.balance(safe_addr).await.map_err(HoprLibError::chain)?;
619            // The key-binding is a persistent on-chain state: if this node was already bound in a
620            // previous run, the fee has already been paid and we can announce even if the safe has
621            // since been drained below the fee. Hence we only wait for funds when the safe is
622            // underfunded *and* the node is not yet key-bound.
623            if safe_hopr_balance < key_binding_fee
624                && chain_api
625                    .await_key_binding(transport_id.public(), Duration::from_secs(1))
626                    .await
627                    .is_err()
628            {
629                tracing::warn!(
630                    %safe_addr,
631                    safe_balance = %safe_hopr_balance,
632                    required = %key_binding_fee,
633                    "the safe does not hold enough wxHOPR to pay the node announcement (key-binding) fee, waiting for \
634                     it to be funded before announcing the node",
635                );
636                let announce_timeout = Duration::from_secs(200);
637                crate::helpers::wait_for_balance(key_binding_fee, announce_timeout, safe_addr, &chain_api)
638                    .await
639                    .map_err(|_| {
640                        HoprLibError::InsufficientFunds(format!(
641                            "the safe {safe_addr} does not hold enough wxHOPR (needs at least {key_binding_fee}) to \
642                             announce the node on chain; fund the safe with wxHOPR and restart the node"
643                        ))
644                    })?;
645            }
646        }
647    }
648
649    let chain_api_clone = chain_api.clone();
650    let me_offchain = *transport_id.public();
651    let node_ready = spawn(async move {
652        chain_api_clone
653            .await_key_binding(&me_offchain, NODE_READY_TIMEOUT)
654            .await
655    });
656
657    tracing::info!(?multiaddresses_to_announce, "announcing node on chain");
658    match chain_api.announce(&multiaddresses_to_announce, &transport_id).await {
659        Ok(awaiter) => {
660            awaiter.await.map_err(|error| {
661                tracing::error!(
662                    ?multiaddresses_to_announce,
663                    %error,
664                    "node announcement failed; this is commonly caused by the safe not holding enough wxHOPR to pay \
665                     the key-binding fee",
666                );
667                HoprLibError::chain(error)
668            })?;
669            tracing::info!(?multiaddresses_to_announce, "node announced successfully");
670        }
671        Err(AnnouncementError::AlreadyAnnounced) => {
672            tracing::info!("node already announced on chain");
673        }
674        Err(error) => {
675            tracing::error!(%error, "failed to transmit node announcement");
676            return Err(HoprLibError::chain(error));
677        }
678    }
679
680    let this_node_account = node_ready
681        .await
682        .map_err(HoprLibError::other)?
683        .map_err(HoprLibError::chain)?;
684    if this_node_account.chain_addr != me_onchain || this_node_account.safe_address.is_none_or(|a| a != safe_addr) {
685        tracing::error!(%this_node_account, "account key-binding mismatch");
686        return Err(HoprLibError::GeneralError("account key-binding mismatch".into()));
687    }
688
689    tracing::info!(%this_node_account, "node account is ready");
690
691    // Network → graph event wiring (subscribe before transport starts)
692    {
693        let network_events = network.subscribe_network_events();
694        let graph_updater = graph.clone();
695        spawn(async move {
696            network_events
697                .for_each(|event| {
698                    let graph_updater = graph_updater.clone();
699                    async move {
700                        let (peer_id, connected) = match event {
701                            hopr_api::network::NetworkEvent::PeerConnected(p) => (p, true),
702                            hopr_api::network::NetworkEvent::PeerDisconnected(p) => (p, false),
703                        };
704                        if let Ok(opk) = hopr_api::OffchainPublicKey::from_peerid(&peer_id) {
705                            graph_updater.record_edge(hopr_api::graph::MeasurableEdge::<
706                                hopr_transport::NeighborTelemetry,
707                                hopr_transport::PathTelemetry,
708                            >::ConnectionStatus {
709                                peer: opk,
710                                connected,
711                            });
712                        } else {
713                            tracing::error!(%peer_id, "failed to convert peer ID to public key for graph update");
714                        }
715                    }
716                })
717                .await;
718        });
719    }
720
721    // Chain → graph event wiring
722    {
723        let chain_events = chain_api
724            .subscribe_with_state_sync([StateSyncOptions::PublicAccounts, StateSyncOptions::OpenedChannels])
725            .map_err(HoprLibError::chain)?;
726
727        let graph_updater = graph.clone();
728        let chain_reader = chain_api.clone();
729
730        let own_chain_addr = me_onchain;
731        let own_packet_key = *transport_id.public();
732
733        let ticket_price = Arc::new(parking_lot::RwLock::new(
734            chain_reader.minimum_ticket_price().await.unwrap_or_default(),
735        ));
736        let win_probability = Arc::new(parking_lot::RwLock::new(
737            chain_reader
738                .minimum_incoming_ticket_win_prob()
739                .await
740                .unwrap_or_default(),
741        ));
742
743        let proc = chain_wiring::process_chain_events(
744            chain_reader,
745            graph_updater,
746            transport_api.surb_store().clone(),
747            chain_events,
748            own_chain_addr,
749            own_packet_key,
750            ticket_price,
751            win_probability,
752            peer_discovery_tx,
753        )
754        .inspect(|_| {
755            tracing::warn!(
756                task = "chain-to-graph event wiring",
757                "long-running background task finished"
758            )
759        });
760        processes.insert(HoprLibProcess::ChannelEvents, hopr_utils::spawn_as_abortable!(proc));
761    }
762
763    Ok(PreHopr {
764        chain_id,
765        transport_id,
766        cfg: ctx.cfg,
767        state: Arc::new(AtomicHoprState::new(HoprState::Uninitialized)),
768        transport_api,
769        chain_api,
770        ticket_event_subscribers: (new_tickets_tx, new_tickets_rx.deactivate()),
771        pix_event_subscribers: (ssa_tx, ssa_rx.deactivate()),
772        processes,
773        session_tx,
774        admission_tx,
775        cover_traffic,
776        network,
777        network_process,
778    })
779}
780
781// ---------------------------------------------------------------------------
782// Build methods — shared via macro to avoid duplicating edge/full logic
783// ---------------------------------------------------------------------------
784
785macro_rules! impl_build_methods {
786    () => {
787        /// Builds an entry [`Hopr`] node.
788        pub async fn build_edge<TFact>(
789            self,
790            ticket_factory: TFact,
791        ) -> Result<Hopr<Chain, Graph, Net, ()>, HoprLibError>
792        where
793            TFact: TicketFactory + Clone + Send + Sync + 'static,
794        {
795            let (configured, session_tx, admission_tx, processes) = self.into_parts();
796            let pre = pre_build_inner(configured, session_tx, admission_tx, processes).await?;
797
798            tracing::info!("starting transport for edge (entry) node");
799            let (socket, transport_processes) = pre
800                .transport_api
801                .run_entry(
802                    pre.cover_traffic,
803                    pre.network,
804                    pre.network_process,
805                    ticket_factory,
806                    Some(BroadcastSenderSink(pre.pix_event_subscribers.0.clone())),
807                )
808                .await?;
809
810            // Drain unrelated packets to avoid missing blackhole
811            spawn(drain_incoming_data(socket.reader()));
812
813            let mut processes = pre.processes;
814            processes.flat_map_extend_from(transport_processes, HoprLibProcess::Transport);
815
816            let hopr = Hopr {
817                chain_id: NodeOnchainIdentity {
818                    node_address: pre.chain_id.public().to_address(),
819                    safe_address: pre.cfg.safe_module.safe_address,
820                    module_address: pre.cfg.safe_module.module_address,
821                },
822                cfg: pre.cfg,
823                state: pre.state.clone(),
824                ticket_event_subscribers: pre.ticket_event_subscribers,
825                pix_event_subscribers: pre.pix_event_subscribers,
826                transport_id: pre.transport_id,
827                transport_api: pre.transport_api,
828                chain_api: pre.chain_api,
829                processes,
830                ticket_manager: (),
831            };
832
833            hopr.state.store(HoprState::Running, std::sync::atomic::Ordering::Relaxed);
834            tracing::info!(
835                id = %hopr.transport_id.public().to_peerid_str(),
836                hopr_lib_version = constants::APP_VERSION,
837                "EDGE NODE STARTED AND RUNNING"
838            );
839
840            Ok(hopr)
841        }
842
843        /// Builds an entry (source) [`Hopr`] node.
844        ///
845        /// Entry nodes do not process tickets, do not have ticket manager state,
846        /// and do not accept incoming sessions.
847        pub async fn build_entry<TFact>(
848            self,
849            ticket_factory: TFact,
850        ) -> Result<Hopr<Chain, Graph, Net, ()>, HoprLibError>
851        where
852            TFact: TicketFactory + Clone + Send + Sync + 'static,
853        {
854            let (configured, session_tx, admission_tx, processes) = self.into_parts();
855            let pre = pre_build_inner(configured, session_tx, admission_tx, processes).await?;
856
857            tracing::info!("starting transport for entry node");
858            let (socket, transport_processes) = pre
859                .transport_api
860                .run_entry(
861                    pre.cover_traffic,
862                    pre.network,
863                    pre.network_process,
864                    ticket_factory,
865                    Some(BroadcastSenderSink(pre.pix_event_subscribers.0.clone())),
866                )
867                .await?;
868
869            // Drain unrelated packets to avoid missing blackhole
870            spawn(drain_incoming_data(socket.reader()));
871
872            let mut processes = pre.processes;
873            processes.flat_map_extend_from(transport_processes, HoprLibProcess::Transport);
874
875            let hopr = Hopr {
876                chain_id: NodeOnchainIdentity {
877                    node_address: pre.chain_id.public().to_address(),
878                    safe_address: pre.cfg.safe_module.safe_address,
879                    module_address: pre.cfg.safe_module.module_address,
880                },
881                cfg: pre.cfg,
882                state: pre.state.clone(),
883                ticket_event_subscribers: pre.ticket_event_subscribers,
884                pix_event_subscribers: pre.pix_event_subscribers,
885                transport_id: pre.transport_id,
886                transport_api: pre.transport_api,
887                chain_api: pre.chain_api,
888                processes,
889                ticket_manager: (),
890            };
891
892            hopr.state.store(HoprState::Running, std::sync::atomic::Ordering::Relaxed);
893            tracing::info!(
894                id = %hopr.transport_id.public().to_peerid_str(),
895                hopr_lib_version = constants::APP_VERSION,
896                "ENTRY NODE STARTED AND RUNNING"
897            );
898
899            Ok(hopr)
900        }
901
902        /// Builds an exit (destination) [`Hopr`] node.
903        ///
904        /// Exit nodes accept incoming sessions and process PIX acknowledgements,
905        /// but do not process tickets (no ticket manager state).
906        pub async fn build_exit<TFact>(
907            self,
908            ticket_factory: TFact,
909        ) -> Result<Hopr<Chain, Graph, Net, ()>, HoprLibError>
910        where
911            TFact: TicketFactory + Clone + Send + Sync + 'static,
912        {
913            let (configured, session_tx, admission_tx, processes) = self.into_parts();
914            let pre = pre_build_inner(configured, session_tx, admission_tx, processes).await?;
915
916            tracing::info!("starting transport for exit node");
917            let (socket, transport_processes) = pre
918                .transport_api
919                .run_exit::<TFact, Ct, _>(
920                    pre.cover_traffic,
921                    pre.network,
922                    pre.network_process,
923                    ticket_factory,
924                    Some(BroadcastSenderSink(pre.pix_event_subscribers.0.clone())),
925                    pre.session_tx,
926                    pre.admission_tx,
927                )
928                .await?;
929
930            // Drain unrelated packets to avoid missing blackhole
931            spawn(drain_incoming_data(socket.reader()));
932
933            let mut processes = pre.processes;
934            processes.flat_map_extend_from(transport_processes, HoprLibProcess::Transport);
935
936            let hopr = Hopr {
937                chain_id: NodeOnchainIdentity {
938                    node_address: pre.chain_id.public().to_address(),
939                    safe_address: pre.cfg.safe_module.safe_address,
940                    module_address: pre.cfg.safe_module.module_address,
941                },
942                cfg: pre.cfg,
943                state: pre.state.clone(),
944                ticket_event_subscribers: pre.ticket_event_subscribers,
945                pix_event_subscribers: pre.pix_event_subscribers,
946                transport_id: pre.transport_id,
947                transport_api: pre.transport_api,
948                chain_api: pre.chain_api,
949                processes,
950                ticket_manager: (),
951            };
952
953            hopr.state.store(HoprState::Running, std::sync::atomic::Ordering::Relaxed);
954            tracing::info!(
955                id = %hopr.transport_id.public().to_peerid_str(),
956                hopr_lib_version = constants::APP_VERSION,
957                "EXIT NODE STARTED AND RUNNING"
958            );
959
960            Ok(hopr)
961        }
962
963        /// Builds a full (relay) [`Hopr`] node.
964        pub async fn build_full<TMgr, TFact>(
965            self,
966            ticket_manager: TMgr,
967            ticket_factory: TFact,
968        ) -> Result<Hopr<Chain, Graph, Net, TMgr>, HoprLibError>
969        where
970            TMgr: TicketManagement + Clone + Send + Sync + 'static,
971            TFact: TicketFactory + Clone + Send + Sync + 'static,
972        {
973            let (configured, session_tx, admission_tx, processes) = self.into_parts();
974            let pre = pre_build_inner(configured, session_tx, admission_tx, processes).await?;
975            let mut processes = pre.processes;
976
977            tracing::info!("starting ticket events processor");
978            let (tickets_tx, tickets_rx) = bounded_sink_channel::<TicketEvent>(8192);
979
980            // Need to use DropAbortable, so that the receiver is dropped when aborted and no new items can be sent by the senders.
981            let (tickets_rx_stream, tickets_handle) = hopr_utils::runtime::DropAbortable::new(tickets_rx);
982
983            processes.insert(HoprLibProcess::TicketEvents, tickets_handle);
984            let new_ticket_tx = pre.ticket_event_subscribers.0.clone();
985            let tmgr_clone = ticket_manager.clone();
986            spawn(
987                hopr_utils::runtime::diagnostics::instrument(
988                    tickets_rx_stream
989                    .for_each(move |event| {
990                        if let TicketEvent::WinningTicket(ticket) = &event
991                            && let Err(error) = tmgr_clone.insert_incoming_ticket(**ticket)
992                        {
993                            tracing::error!(%error, "failed to insert incoming ticket");
994                        }
995                        if let Err(error) = new_ticket_tx.try_broadcast(event) {
996                            tracing::error!(%error, "failed to broadcast ticket event");
997                        }
998                        futures::future::ready(())
999                    })
1000                    .inspect(|_| {
1001                        tracing::warn!(task = %HoprLibProcess::TicketEvents, "long-running background task finished")
1002                    }),
1003                    "hopr_lib_ticket_events",
1004                    module_path!(),
1005                    file!(),
1006                    line!(),
1007                ),
1008            );
1009
1010            {
1011                let chain_for_neglect = pre.chain_api.clone();
1012                let tmgr_for_neglect = ticket_manager.clone();
1013                let events = pre.chain_api.subscribe().map_err(HoprLibError::chain)?;
1014                let (neglect_handle, neglect_reg) = hopr_utils::runtime::AbortHandle::new_pair();
1015                let neglect_task = futures::stream::Abortable::new(
1016                    events.filter_map(move |event| {
1017                        futures::future::ready(match event {
1018                            ChainEvent::ChannelClosed(ch) => Some(ch),
1019                            _ => None,
1020                        })
1021                    }),
1022                    neglect_reg,
1023                )
1024                .for_each(move |closed_channel| {
1025                    let chain = chain_for_neglect.clone();
1026                    let tmgr = tmgr_for_neglect.clone();
1027                    async move {
1028                        match closed_channel.direction(chain.me()) {
1029                            Some(ChannelDirection::Incoming) => {
1030                                match tmgr.neglect_tickets(closed_channel.get_id(), None) {
1031                                    Ok(neglected) if !neglected.is_empty() => {
1032                                        tracing::warn!(
1033                                            num_neglected = neglected.len(),
1034                                            %closed_channel,
1035                                            "tickets on incoming closed channel were neglected"
1036                                        );
1037                                    }
1038                                    Ok(_) => {}
1039                                    Err(error) => {
1040                                        tracing::error!(
1041                                            %error, %closed_channel,
1042                                            "failed to neglect tickets on closed channel"
1043                                        );
1044                                    }
1045                                }
1046                            }
1047                            Some(ChannelDirection::Outgoing) => {}
1048                            _ => {}
1049                        }
1050                    }
1051                })
1052                .inspect(|_| {
1053                    tracing::warn!(
1054                        task = %HoprLibProcess::ChannelClosureNeglect,
1055                        "channel closure ticket neglect task finished"
1056                    )
1057                });
1058                spawn(hopr_utils::runtime::diagnostics::instrument(
1059                    neglect_task,
1060                    "hopr_lib_channel_closure_neglect",
1061                    module_path!(),
1062                    file!(),
1063                    line!(),
1064                ));
1065                processes.insert(HoprLibProcess::ChannelClosureNeglect, neglect_handle);
1066            }
1067
1068            tracing::info!("starting transport for full (relay) node");
1069            let (socket, transport_processes) = pre
1070                .transport_api
1071                .run_relay(
1072                    pre.cover_traffic,
1073                    pre.network,
1074                    pre.network_process,
1075                    tickets_tx,
1076                    ticket_factory,
1077                    Some(BroadcastSenderSink(pre.pix_event_subscribers.0.clone())),
1078                    pre.session_tx,
1079                    pre.admission_tx,
1080                )
1081                .await?;
1082            // Drain unrelated packets to avoid missing blackhole
1083            spawn(drain_incoming_data(socket.reader()));
1084            processes.flat_map_extend_from(transport_processes, HoprLibProcess::Transport);
1085
1086            let hopr = Hopr {
1087                chain_id: NodeOnchainIdentity {
1088                    node_address: pre.chain_id.public().to_address(),
1089                    safe_address: pre.cfg.safe_module.safe_address,
1090                    module_address: pre.cfg.safe_module.module_address,
1091                },
1092                cfg: pre.cfg,
1093                state: pre.state.clone(),
1094                ticket_event_subscribers: pre.ticket_event_subscribers,
1095                pix_event_subscribers: pre.pix_event_subscribers,
1096                transport_id: pre.transport_id,
1097                transport_api: pre.transport_api,
1098                chain_api: pre.chain_api,
1099                processes,
1100                ticket_manager,
1101            };
1102
1103            hopr.state.store(HoprState::Running, std::sync::atomic::Ordering::Relaxed);
1104
1105            tracing::info!(
1106                id = %hopr.transport_id.public().to_peerid_str(),
1107                hopr_lib_version = constants::APP_VERSION,
1108                "FULL NODE STARTED AND RUNNING"
1109            );
1110
1111            #[cfg(all(feature = "telemetry", not(test)))]
1112            METRIC_HOPR_NODE_INFO.set(
1113                &[
1114                    &hopr.transport_id.public().to_peerid_str(),
1115                    &hopr.chain_id.node_address.to_string(),
1116                    &hopr.chain_id.safe_address.to_string(),
1117                    &hopr.chain_id.module_address.to_string(),
1118                ],
1119                1.0,
1120            );
1121
1122            Ok(hopr)
1123        }
1124    };
1125}
1126
1127// When session-server is ON: build methods only on HoprBuilderWithSession
1128#[cfg(feature = "session-server")]
1129impl<Chain, Graph, Net, Ct> HoprBuilderWithSession<Chain, Graph, Net, Ct>
1130where
1131    Chain: HoprChainApi + Clone + Send + Sync + 'static,
1132    Graph: HoprGraphApi<HoprNodeId = hopr_api::OffchainPublicKey> + Clone + Send + Sync + 'static,
1133    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed:
1134        hopr_api::graph::traits::EdgeObservableRead + Send + 'static,
1135    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
1136    Net: NetworkView + NetworkStreamControl + Send + Sync + Clone + 'static,
1137    Ct: ProbingTrafficGeneration + CoverTrafficGeneration + Send + Sync + 'static,
1138{
1139    impl_build_methods!();
1140
1141    fn into_parts(self) -> BuildParts<Chain, Graph, Net, Ct> {
1142        let mut processes = AbortableList::<HoprLibProcess>::default();
1143        processes.insert(HoprLibProcess::SessionServer, self.session_handle);
1144        processes.insert(HoprLibProcess::SessionAdmission, self.admission_handle);
1145        (self.inner, self.session_tx, Some(self.admission_tx), processes)
1146    }
1147}
1148
1149// When session-server is OFF: build methods directly on HoprBuilderConfigured
1150#[cfg(not(feature = "session-server"))]
1151impl<Chain, Graph, Net, Ct> HoprBuilderConfigured<Chain, Graph, Net, Ct>
1152where
1153    Chain: HoprChainApi + Clone + Send + Sync + 'static,
1154    Graph: HoprGraphApi<HoprNodeId = hopr_api::OffchainPublicKey> + Clone + Send + Sync + 'static,
1155    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed:
1156        hopr_api::graph::traits::EdgeObservableRead + Send + 'static,
1157    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
1158    Net: NetworkView + NetworkStreamControl + Send + Sync + Clone + 'static,
1159    Ct: ProbingTrafficGeneration + CoverTrafficGeneration + Send + Sync + 'static,
1160{
1161    impl_build_methods!();
1162
1163    fn into_parts(self) -> BuildParts<Chain, Graph, Net, Ct> {
1164        let (tx, _rx) = futures::channel::mpsc::channel::<IncomingSession>(1);
1165        let processes = AbortableList::<HoprLibProcess>::default();
1166        (self, tx, None, processes)
1167    }
1168}
hopr_lib/builder.rs

hopr_lib/
builder.rs
