hopr_transport/

lib.rs

//! The crate aggregates and composes individual transport level objects and functionality
//! into a unified [`crate::HoprTransport`] object with the goal of isolating the transport layer
//! and defining a fully specified transport API.
//!
//! See also the `hopr_protocol_start` crate for details on Start sub-protocol which initiates a Session.
//!
//! As such, the transport layer components should be only those that are directly necessary to:
//!
//! 1. send and receive a packet, acknowledgement or ticket aggregation request
//! 2. send and receive a network telemetry request
//! 3. automate transport level processes
//! 4. algorithms associated with the transport layer operational management
//! 5. interface specifications to allow modular behavioral extensions

/// Configuration of the [crate::HoprTransport].
pub mod config;
/// Constants used and exposed by the crate.
pub mod constants;
/// Errors used by the crate.
pub mod errors;
/// Graph-based path planning for the HOPR transport layer.
pub mod path;
/// Transport binary protocol layer (codec, pipeline, heartbeat, stream).
pub mod protocol;
/// Test utilities: emulated peer wiring, stub chain API, shared keypair/payload fixtures.
/// Enabled by the `testing` feature; every item is a zero-cost no-op stub when absent.
#[cfg(feature = "testing")]
pub mod testing;

mod multiaddrs;

#[cfg(feature = "capture")]
mod capture;
mod pipeline;
pub mod socket;

use std::{
    sync::{Arc, OnceLock},
    time::Duration,
};

use constants::MAXIMUM_MSG_OUTGOING_BUFFER_SIZE;
use futures::{FutureExt, SinkExt, StreamExt, channel::mpsc::Sender, stream::select_with_strategy};
use futures_concurrency::stream::StreamExt as ConcurrentStreamExt;
pub use hopr_api::{
    Multiaddr, PeerId,
    network::{Health, traits::NetworkView},
    types::{
        crypto::{
            keypairs::{ChainKeypair, Keypair, OffchainKeypair},
            types::{HalfKeyChallenge, Hash, OffchainPublicKey, SimplePseudonym},
        },
        internal::{prelude::HoprPseudonym, routing::RoutingOptions},
    },
};
use hopr_api::{
    chain::{ChainKeyOperations, ChainReadAccountOperations, ChainReadChannelOperations, ChainValues},
    ct::{CoverTrafficGeneration, ProbingTrafficGeneration},
    graph::{NetworkGraphUpdate, NetworkGraphView, traits::EdgeObservableRead},
    network::{BoxedProcessFn, NetworkStreamControl},
    types::primitive::prelude::*,
};
pub use hopr_crypto_packet::HoprPixSpec;
pub use hopr_protocol_app::prelude::{ApplicationData, ApplicationDataIn, ApplicationDataOut, Tag};
pub use hopr_protocol_hopr::{MemorySurbStore, SurbStore};
/// Re-exported so a consumer can name `<HoprPixSpec as PixSpec>::DepositAddress` and assert at
/// compile time that it is the variant it can actually settle. Which instantiation is in play is
/// a feature-graph outcome rather than a local decision, and getting it wrong is otherwise silent
/// until deposits stop happening at runtime.
pub use hopr_protocol_pix::PixSpec;
pub use hopr_protocol_pix::RecoveredSsa;
use hopr_protocol_pix::ShareResolution;
pub use hopr_transport_probe::{NeighborTelemetry, PathTelemetry, errors::ProbeError, ping::PingQueryReplier};
use hopr_transport_probe::{
    Probe,
    ping::{PingConfig, Pinger},
};
pub use hopr_transport_session as session;
#[cfg(feature = "runtime-tokio")]
pub use hopr_transport_session::transfer_session;
use hopr_transport_session::{
    AgreedSsaQuota, DispatchResult, HoprSessionInPixEvent, HoprSessionOutPixEvent, PixToolbox, SessionManager,
    SessionManagerConfig,
};
pub use hopr_transport_session::{
    Capabilities as SessionCapabilities, Capability as SessionCapability, FlowControlConfig, HoprSession,
    IncomingSession, InvalidPixParams, LOCAL_PIX_SUITE, PixParams, SESSION_MTU, SURB_SIZE, ServiceId,
    SessionClientConfig, SessionId, SessionTarget, SurbBalancerConfig,
    errors::{SessionManagerError, TransportSessionError},
};
#[cfg(feature = "telemetry")]
pub use hopr_transport_session::{SessionAckMode, SessionLifecycleState};
pub use hopr_transport_tag_allocator::TagAllocatorConfig;
use hopr_utils::{
    network_types::{
        crossfire_sink::{CrossfireSink, bounded_sink_channel},
        prelude::*,
    },
    runtime::AbortableList,
};
pub use multiaddr::Protocol;
use tracing::{debug, warn};

#[cfg(feature = "runtime-tokio")]
use crate::path::BackgroundPathCacheRefreshable;
pub use crate::{config::HoprProtocolConfig, protocol::PeerProtocolCounterRegistry};
use crate::{
    config::PixGlobalConfig,
    constants::SESSION_INITIATION_TIMEOUT_BASE,
    errors::HoprTransportError,
    multiaddrs::strip_p2p_protocol,
    path::{HoprGraphPathSelector, PathPlanner},
    pipeline::HoprPacketPipelineBuilder,
    socket::HoprSocket,
};

pub const APPLICATION_TAG_RANGE: std::ops::Range<Tag> = Tag::APPLICATION_TAG_RANGE;

pub use hopr_api as api;
use hopr_api::{
    chain::{ChainReadTicketOperations, ChainWriteTicketOperations, PixDepositSecret},
    node::{PixDepositAddressReceived, PixEvent, PixNewDepositAddress, PixPrivateKeyRecovered},
    tickets::TicketFactory,
    types::internal::routing::DestinationRouting,
};
use hopr_crypto_packet::HoprShareResolution;
use rust_stream_ext_concurrent::then_concurrent::StreamThenConcurrentExt;

// Needs lazy-static, since Duration multiplication by a constant is yet not a const-operation.
lazy_static::lazy_static! {
    static ref SESSION_INITIATION_TIMEOUT_MAX: Duration = 2 * SESSION_INITIATION_TIMEOUT_BASE * RoutingOptions::MAX_INTERMEDIATE_HOPS as u32;

    static ref PEER_ID_CACHE: moka::sync::Cache<PeerId, OffchainPublicKey> = moka::sync::Cache::builder()
        .time_to_idle(Duration::from_mins(15))
        .max_capacity(10_000)
        .build();

    static ref RANDOM_DATA: [u8; 400] = hopr_api::types::crypto_random::random_bytes();
}

/// Fraction of the SURB ring buffer a Session's balancer target may occupy.
///
/// The remainder is headroom for balancer overshoot. The target used to be the ring buffer capacity
/// itself, which left none: a Session whose buffer sat at target — the normal state when Entry → Exit
/// traffic far exceeds the reverse, since the Exit then drains almost nothing — turned every
/// overshot SURB into an immediate eviction of the oldest entry. Under PIX that is a lost SSA share,
/// not just a lost SURB, because a share reaches the reconstructor only when its SURB is used.
const SURB_BUFFER_TARGET_NUMERATOR: usize = 2;
const SURB_BUFFER_TARGET_DENOMINATOR: usize = 3;

/// Largest SURB buffer target a Session may request, given the ring buffer capacity backing it.
///
/// Kept strictly below `rb_capacity` so the balancer has somewhere to overshoot into before the
/// ring buffer starts overwriting. See [`SURB_BUFFER_TARGET_NUMERATOR`].
const fn surb_buffer_target_ceiling(rb_capacity: usize) -> usize {
    // Saturating rather than wrapping: an operator-supplied capacity is only range-validated at the
    // low end, and the multiplication would otherwise overflow on an absurd value.
    rb_capacity.saturating_mul(SURB_BUFFER_TARGET_NUMERATOR) / SURB_BUFFER_TARGET_DENOMINATOR
}

/// PeerId -> OffchainPublicKey is a CPU-intensive blocking operation.
///
/// This helper uses a cached static object to speed up the lookup and avoid blocking the async
/// runtime on repeated conversions for the same [`PeerId`]s.
pub fn peer_id_to_public_key(peer_id: &PeerId) -> crate::errors::Result<OffchainPublicKey> {
    PEER_ID_CACHE
        .try_get_with_by_ref(peer_id, move || {
            OffchainPublicKey::from_peerid(peer_id).map_err(|e| e.into())
        })
        .map_err(|e: Arc<HoprTransportError>| {
            crate::errors::HoprTransportError::Other(anyhow::anyhow!(
                "failed to convert peer_id ({:?}) to an offchain public key: {e}",
                peer_id
            ))
        })
}

#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, strum::Display)]
pub enum HoprTransportProcess {
    #[strum(to_string = "component responsible for the transport medium (libp2p swarm)")]
    Medium,
    #[strum(to_string = "HOPR packet pipeline ({0})")]
    Pipeline(protocol::PacketPipelineProcesses),
    #[strum(to_string = "session manager sub-process #{0}")]
    SessionsManagement(usize),
    #[strum(to_string = "network probing sub-process: {0}")]
    Probing(hopr_transport_probe::HoprProbeProcess),
    #[cfg(feature = "runtime-tokio")]
    #[strum(to_string = "path cache refresh")]
    PathRefresh,
    #[strum(to_string = "pix protocol event transformation")]
    PixEvents,
    #[strum(to_string = "sync of outgoing ticket indices")]
    OutgoingIndexSync,
    #[strum(to_string = "periodic protocol counter flush")]
    CounterFlush,
    /// Periodically reports SURB round-trip counts into the network graph.
    SurbFlush,
    #[strum(to_string = "mixer→wire forwarder")]
    MixerForwarder,
    #[cfg(feature = "capture")]
    #[strum(to_string = "packet capture")]
    Capture,
}

/// HOPR protocol specific instantiation of the SessionManager.
type HoprSessionManager = SessionManager<CrossfireSink<(DestinationRouting, ApplicationDataOut)>>;

/// Allows configuration of one specific [`HoprSession`].
///
/// The configurator does not prevent the Session from being closed
/// or the Session manager from being dropped.
#[derive(Debug, Clone)]
pub struct HoprSessionConfigurator {
    id: SessionId,
    // Makes sure configurator does not extend lifetime of the SessionManager.
    smgr: std::sync::Weak<HoprSessionManager>,
}

impl HoprSessionConfigurator {
    /// [`SessionId`] of the session this object can configure.
    pub fn id(&self) -> &SessionId {
        &self.id
    }

    /// Sends a Session Keep-Alive packet over the Session.
    ///
    /// NOTE: This usually carries at least 2 SURBs on the HOPR protocol level and can be
    /// used for manual SURB balancing.
    ///
    /// NOTE: This operation only sends the Session Keep-Alive packet and **DOES NOT** guarantee the other side
    /// has received it.
    pub async fn ping(&self) -> errors::Result<()> {
        Ok(self
            .smgr
            .upgrade()
            .ok_or(HoprTransportError::Other(anyhow::anyhow!("session manager is dropped")))?
            .ping_session(&self.id)
            .await?)
    }

    /// Gets the configuration of the SURB balancer.
    ///
    /// Returns an error if the Session is closed, the Session manager is gone.
    ///
    /// Returns `Ok(None)` if the Session has been created without a SURB balancer.
    pub fn get_surb_balancer_config(&self) -> errors::Result<Option<SurbBalancerConfig>> {
        Ok(self
            .smgr
            .upgrade()
            .ok_or(HoprTransportError::Other(anyhow::anyhow!("session manager is dropped")))?
            .get_surb_balancer_config(&self.id)?)
    }

    /// Updates the configuration of the SURB balancer.
    ///
    /// Returns an error if the Session is closed, the Session manager is gone, or the
    /// Session has been created without a SURB balancer.
    pub fn update_surb_balancer_config(&self, config: SurbBalancerConfig) -> errors::Result<()> {
        Ok(self
            .smgr
            .upgrade()
            .ok_or(HoprTransportError::Other(anyhow::anyhow!("session manager is dropped")))?
            .update_surb_balancer_config(&self.id, config)?)
    }

    /// Explicitly closes the underlying Session in the [`SessionManager`].
    ///
    /// Returns `true` if the session was found and closed, `false` if it was
    /// already gone (or the manager is dropped). Frees the per-session state
    /// (frame reassembly buffers, control channels, …) immediately rather than
    /// waiting for the manager's idle-timeout eviction.
    pub fn close(&self) -> bool {
        match self.smgr.upgrade() {
            Some(smgr) => smgr.close_session(&self.id),
            None => false,
        }
    }
}

/// Interface into the physical transport mechanism allowing all off-chain HOPR-related tasks on
/// the transport.
pub struct HoprTransport<Chain, Graph, Net> {
    packet_key: OffchainKeypair,
    chain_key: ChainKeypair,
    chain_api: Chain,
    ping: Arc<OnceLock<Pinger>>,
    network: Arc<OnceLock<Net>>,
    graph: Graph,
    path_planner: PathPlanner<MemorySurbStore, Chain, HoprGraphPathSelector<Graph>>,
    my_multiaddresses: Vec<Multiaddr>,
    smgr: Arc<HoprSessionManager>,
    session_telemetry_tag_allocator: Arc<dyn hopr_transport_tag_allocator::TagAllocator + Send + Sync>,
    probing_tag_allocator: Arc<dyn hopr_transport_tag_allocator::TagAllocator + Send + Sync>,
    counters: PeerProtocolCounterRegistry,
    cfg: HoprProtocolConfig,
}

impl<Chain, Graph, Net> HoprTransport<Chain, Graph, Net>
where
    Chain: ChainReadChannelOperations
        + ChainReadAccountOperations
        + ChainWriteTicketOperations
        + ChainKeyOperations
        + ChainReadTicketOperations
        + ChainValues
        + Clone
        + Send
        + Sync
        + 'static,
    Graph: NetworkGraphView<NodeId = OffchainPublicKey>
        + NetworkGraphUpdate
        + hopr_api::graph::NetworkGraphWrite<NodeId = OffchainPublicKey>
        + hopr_api::graph::NetworkGraphTraverse<NodeId = OffchainPublicKey>
        + Clone
        + Send
        + Sync
        + 'static,
    <Graph as NetworkGraphView>::Observed: hopr_api::graph::traits::EdgeObservableRead + Send,
    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed:
        hopr_api::graph::traits::EdgeObservableRead + Send + 'static,
    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
    Net: NetworkView + NetworkStreamControl + Clone + Send + Sync + 'static,
{
    pub fn new(
        identity: (&ChainKeypair, &OffchainKeypair),
        resolver: Chain,
        graph: Graph,
        my_multiaddresses: Vec<Multiaddr>,
        cfg: HoprProtocolConfig,
    ) -> errors::Result<Self> {
        let me_offchain = *identity.1.public();
        let planner_config = cfg.path_planner;
        let selector = HoprGraphPathSelector::new(
            me_offchain,
            graph.clone(),
            planner_config.max_cached_paths,
            planner_config.edge_penalty,
            planner_config.min_ack_rate,
            planner_config.min_paths_anonymity_floor,
        );

        let tag_allocators = hopr_transport_tag_allocator::create_allocators_from_config(&cfg.session.tag_allocator)?;

        let mut session_telemetry_tag_allocator = None;
        let mut probing_tag_allocator = None;
        for (usage, alloc) in tag_allocators {
            match usage {
                // TODO: cleanup of Session tag allocators needed * (#8199)
                hopr_transport_tag_allocator::Usage::Session => {}
                hopr_transport_tag_allocator::Usage::SessionTerminalTelemetry => {
                    session_telemetry_tag_allocator = Some(alloc)
                }
                hopr_transport_tag_allocator::Usage::ProvingTelemetry => probing_tag_allocator = Some(alloc),
            }
        }
        let session_telemetry_tag_allocator = session_telemetry_tag_allocator
            .ok_or_else(|| HoprTransportError::Api("session telemetry tag allocator missing".into()))?;
        let probing_tag_allocator =
            probing_tag_allocator.ok_or_else(|| HoprTransportError::Api("probing tag allocator missing".into()))?;

        // A pseudonym's SURB ring buffer is dropped once no SURBs have arrived for
        // `pseudonyms_lifetime` (600 s by default), which is the point at which that pseudonym's
        // return path becomes permanently unresolvable. Nothing else in the node can be made to
        // reach that state on a test timescale — the Session slot is evicted for idleness long
        // before it — so without a way to shorten this timer, the behaviour past it is not
        // observable end-to-end at all. Floored at the same `MINIMUM_SURB_LIFETIME` the config
        // validator enforces, so this cannot reach a value the config file could not also express.
        let surb_store_cfg = hopr_protocol_hopr::SurbStoreConfig {
            pseudonyms_lifetime: std::env::var("HOPR_INTERNAL_SURB_PSEUDONYM_LIFETIME_MS")
                .ok()
                .and_then(|s| s.trim().parse::<u64>().ok())
                .map(Duration::from_millis)
                .map(|d| d.max(hopr_protocol_hopr::MINIMUM_SURB_LIFETIME))
                .unwrap_or(cfg.packet.surb_store.pseudonyms_lifetime),
            ..cfg.packet.surb_store
        };

        // Built before the session manager so the latter can be handed the seam that lets a
        // session re-plan its return path on sustained loss.
        let path_planner = PathPlanner::new(
            me_offchain,
            MemorySurbStore::new(surb_store_cfg),
            resolver.clone(),
            selector,
            planner_config,
        );

        Ok(Self {
            packet_key: identity.1.clone(),
            chain_key: identity.0.clone(),
            ping: Arc::new(OnceLock::new()),
            network: Arc::new(OnceLock::new()),
            graph,
            path_planner,
            my_multiaddresses,
            smgr: Arc::new(SessionManager::new(SessionManagerConfig {
                frame_mtu: std::env::var("HOPR_SESSION_FRAME_SIZE")
                    .ok()
                    .and_then(|s| s.parse::<usize>().ok())
                    .unwrap_or_else(|| SessionManagerConfig::default().frame_mtu)
                    .max(SESSION_MTU),
                max_frame_timeout: std::env::var("HOPR_SESSION_FRAME_TIMEOUT_MS")
                    .ok()
                    .and_then(|s| s.parse::<u64>().ok().map(Duration::from_millis))
                    .unwrap_or_else(|| SessionManagerConfig::default().max_frame_timeout)
                    .max(Duration::from_millis(100)),
                max_frames_behind_gap: cfg.session.max_frames_behind_gap,
                max_buffered_segments: std::env::var("HOPR_SESSION_MAX_BUFFERED_SEGMENTS")
                    .ok()
                    .and_then(|s| s.parse::<usize>().ok())
                    .unwrap_or_else(|| SessionManagerConfig::default().max_buffered_segments),
                initiation_timeout_base: SESSION_INITIATION_TIMEOUT_BASE,
                idle_timeout: cfg.session.idle_timeout,
                balancer_sampling_interval: cfg.session.balancer_sampling_interval,
                initial_return_session_egress_rate: 10,
                minimum_surb_buffer_duration: cfg.session.balancer_minimum_surb_buffer_duration,
                maximum_surb_buffer_size: surb_buffer_target_ceiling(cfg.packet.surb_store.rb_capacity),
                // The lower bound is enforced once in `SessionManager::new` via
                // `MIN_SURB_BUFFER_NOTIFICATION_PERIOD`; don't duplicate that floor as a literal here.
                surb_balance_notify_period: std::env::var("HOPR_SESSION_SURB_BALANCE_NOTIFY_PERIOD_MS")
                    .ok()
                    .and_then(|s| s.parse::<u64>().ok())
                    .map(|ms| Some(Duration::from_millis(ms)))
                    .unwrap_or(cfg.session.surb_balance_notify_period),
                surb_target_notify: true,
                maximum_sessions: cfg.session.maximum_managed_sessions,
                pix_config: cfg.incoming_session_pix_config.clone(),
                max_ssas_per_ssa_request: cfg.pix.max_ssas_per_request,
                ..Default::default()
            })),
            chain_api: resolver,
            session_telemetry_tag_allocator,
            probing_tag_allocator,
            counters: PeerProtocolCounterRegistry::default(),
            cfg,
        })
    }

    /// Execute all processes of the [`HoprTransport`] object as a **Relay** node.
    ///
    /// Relay nodes run the full packet pipeline including incoming ticket/acknowledgement
    /// processing and require a [`futures::Sink`] for ticket events as well as an
    /// `on_incoming_session` channel from the SessionManager (they can accept incoming sessions).
    ///
    /// The Relay node may also opt in to allow itself to use the PIX protocol by setting
    /// the `exit_ack_share` if it wishes to also act as an Exit node.
    #[allow(clippy::too_many_arguments)]
    pub async fn run_relay<T, TFact, Ct, PixEvt>(
        &self,
        cover_traffic: Ct,
        network: Net,
        network_process: BoxedProcessFn,
        ticket_events: T,
        ticket_factory: TFact,
        exit_ack_share: Option<PixEvt>,
        on_incoming_session: Sender<IncomingSession>,
    ) -> errors::Result<(
        HoprSocket<
            futures::stream::BoxStream<'static, ApplicationDataIn>,
            CrossfireSink<(DestinationRouting, ApplicationDataOut)>,
        >,
        AbortableList<HoprTransportProcess>,
    )>
    where
        T: futures::Sink<hopr_api::node::TicketEvent> + Clone + Send + Unpin + 'static,
        T::Error: std::error::Error + Clone + Send,
        Ct: ProbingTrafficGeneration + CoverTrafficGeneration + Send + Sync + 'static,
        TFact: TicketFactory + Clone + Send + Sync + 'static,
        PixEvt: futures::Sink<PixEvent> + Clone + Unpin + Send + 'static,
        PixEvt::Error: std::error::Error + Clone + Sync + Send + 'static,
    {
        self.run_inner(
            protocol::NodeType::Relay,
            cover_traffic,
            network,
            network_process,
            ticket_events,
            ticket_factory,
            exit_ack_share,
            Some(on_incoming_session),
        )
        .await
    }

    /// Execute all processes of the [`HoprTransport`] object as an **Exit** (destination) node.
    ///
    /// Exit nodes do not process tickets but keep the incoming acknowledgement
    /// pipeline running and can accept incoming sessions via SessionManager.
    ///
    /// The Exit nodes also work with the PIX protocol, so they process incoming acknowledgements
    /// to decrypt PIX shares.
    pub async fn run_exit<TFact, Ct, PixEvt>(
        &self,
        cover_traffic: Ct,
        network: Net,
        network_process: BoxedProcessFn,
        ticket_factory: TFact,
        pix_events: Option<PixEvt>,
        on_incoming_session: Sender<IncomingSession>,
    ) -> errors::Result<(
        HoprSocket<
            futures::stream::BoxStream<'static, ApplicationDataIn>,
            CrossfireSink<(DestinationRouting, ApplicationDataOut)>,
        >,
        AbortableList<HoprTransportProcess>,
    )>
    where
        Ct: ProbingTrafficGeneration + CoverTrafficGeneration + Send + Sync + 'static,
        TFact: TicketFactory + Clone + Send + Sync + 'static,
        PixEvt: futures::Sink<PixEvent> + Clone + Unpin + Send + 'static,
        PixEvt::Error: std::error::Error + Clone + Sync + Send + 'static,
    {
        self.run_inner(
            protocol::NodeType::Exit,
            cover_traffic,
            network,
            network_process,
            futures::sink::drain(),
            ticket_factory,
            pix_events,
            Some(on_incoming_session),
        )
        .await
    }

    /// Execute all processes of the [`HoprTransport`] object as an **Entry** (source) node.
    ///
    /// Entry nodes do not process tickets, do not start the incoming acknowledgement
    /// pipeline, and do not accept incoming sessions — therefore, they require neither a
    /// `ticket_events` sink nor an `on_incoming_session` channel.
    pub async fn run_entry<TFact, Ct, PixEvt>(
        &self,
        cover_traffic: Ct,
        network: Net,
        network_process: BoxedProcessFn,
        ticket_factory: TFact,
        pix_events: Option<PixEvt>,
    ) -> errors::Result<(
        HoprSocket<
            futures::stream::BoxStream<'static, ApplicationDataIn>,
            CrossfireSink<(DestinationRouting, ApplicationDataOut)>,
        >,
        AbortableList<HoprTransportProcess>,
    )>
    where
        Ct: ProbingTrafficGeneration + CoverTrafficGeneration + Send + Sync + 'static,
        TFact: TicketFactory + Clone + Send + Sync + 'static,
        PixEvt: futures::Sink<PixEvent> + Clone + Unpin + Send + 'static,
        PixEvt::Error: std::error::Error + Clone + Sync + Send + 'static,
    {
        self.run_inner(
            protocol::NodeType::Entry,
            cover_traffic,
            network,
            network_process,
            futures::sink::drain(),
            ticket_factory,
            pix_events,
            None,
        )
        .await
    }

    /// Internal worker driving all node-type variants of `HoprTransport::run_*`.
    ///
    /// Branches on `role`:
    /// - [`protocol::NodeType::Relay`]: full packet pipeline + SessionManager.
    /// - [`protocol::NodeType::Exit`]: ack-drain pipeline + incoming Sessions.
    /// - [`protocol::NodeType::Entry`]: no ack pipeline, no incoming Sessions.
    #[allow(clippy::too_many_arguments)]
    async fn run_inner<T, TFact, Ct, PixEvt>(
        &self,
        role: protocol::NodeType,
        cover_traffic: Ct,
        network: Net,
        network_process: BoxedProcessFn,
        ticket_events: T,
        ticket_factory: TFact,
        exit_ack_share: Option<PixEvt>,
        on_incoming_session: Option<Sender<IncomingSession>>,
    ) -> errors::Result<(
        HoprSocket<
            futures::stream::BoxStream<'static, ApplicationDataIn>,
            CrossfireSink<(DestinationRouting, ApplicationDataOut)>,
        >,
        AbortableList<HoprTransportProcess>,
    )>
    where
        T: futures::Sink<hopr_api::node::TicketEvent> + Clone + Send + Unpin + 'static,
        T::Error: std::error::Error + Clone + Send,
        Ct: ProbingTrafficGeneration + CoverTrafficGeneration + Send + Sync + 'static,
        TFact: TicketFactory + Clone + Send + Sync + 'static,
        PixEvt: futures::Sink<PixEvent> + Clone + Unpin + Send + 'static,
        PixEvt::Error: std::error::Error + Clone + Sync + Send + 'static,
    {
        let mut processes = AbortableList::<HoprTransportProcess>::default();

        let (unresolved_routing_msg_tx, unresolved_routing_msg_rx) =
            bounded_sink_channel::<(DestinationRouting, ApplicationDataOut)>(MAXIMUM_MSG_OUTGOING_BUFFER_SIZE);

        // -- transport medium

        let transport_network = network;
        let transport_layer_process = network_process;

        let msg_codec = crate::protocol::HoprBinaryCodec {};
        let (wire_msg_tx, wire_msg_rx) =
            protocol::stream::process_stream_protocol(msg_codec, transport_network.clone(), self.cfg.stream).await?;

        // Shared mixing channel: all per-destination clones of `mixing_channel_tx` push into one
        // heap, so cross-destination packets are mixed together rather than each destination
        // getting its own independent delay queue. The single forwarder task owns the receiver
        // (and therefore the heap timer) — no per-clone waker coordination is needed.
        let mut mixer_cfg = self.cfg.mixer;
        mixer_cfg.metric_delay_window = u64::try_from(5 * mixer_cfg.delay_range.as_millis())
            .unwrap_or(u64::MAX)
            .max(1);
        let (mixing_channel_tx, mix_rx) = hopr_transport_mixer::channel(mixer_cfg);
        let transit_latency_cfg = self.cfg.transit_latency;
        processes.insert(
            HoprTransportProcess::MixerForwarder,
            hopr_utils::spawn_as_abortable!(async move {
                let mut mix_rx = mix_rx;
                let mut wire_sink = wire_msg_tx;

                if let Some(lat) = transit_latency_cfg {
                    // Concurrent transit-latency fan-out: spawn one short-lived task per
                    // packet so each packet ages through its own ~mean delay independently.
                    // A burst of N packets at mean=50 ms takes ~50 ms total, not N×50 ms.
                    //
                    // Without a tokio runtime the latency is silently skipped (pass-through).
                    #[cfg(feature = "runtime-tokio")]
                    {
                        let (tx, rx) = futures::channel::mpsc::unbounded();
                        let fan_out = async move {
                            while let Some(item) = futures::StreamExt::next(&mut mix_rx).await {
                                let mean_us = lat.mean.as_micros() as f64;
                                let std_us = lat.std_dev.as_micros() as f64;
                                let delay_us = if std_us > 0.0 {
                                    use rand_distr::{Distribution, Normal};
                                    Normal::new(mean_us, std_us)
                                        .expect("transit latency Normal params are valid")
                                        .sample(&mut rand::rng())
                                        .max(0.0_f64)
                                } else {
                                    mean_us.max(0.0)
                                };
                                let delay = Duration::from_micros(delay_us as u64);
                                let item_tx = tx.clone();
                                hopr_utils::runtime::prelude::spawn(async move {
                                    if !delay.is_zero() {
                                        futures_timer::Delay::new(delay).await;
                                    }
                                    let _ = item_tx.unbounded_send(item);
                                });
                            }
                            // `tx` drops here; the channel closes once all per-packet tasks send
                        };
                        let fan_in = async move {
                            let mut rx = rx;
                            while let Some(item) = futures::StreamExt::next(&mut rx).await {
                                if wire_sink.send(item).await.is_err() {
                                    tracing::error!(
                                        task = %HoprTransportProcess::MixerForwarder,
                                        "wire sink dropped — discarding transit-delayed packet"
                                    );
                                    break;
                                }
                            }
                        };
                        futures::join!(fan_out, fan_in);
                    }
                    #[cfg(not(feature = "runtime-tokio"))]
                    {
                        let _ = lat;
                        while let Some(item) = futures::StreamExt::next(&mut mix_rx).await {
                            if wire_sink.send(item).await.is_err() {
                                tracing::error!(
                                    task = %HoprTransportProcess::MixerForwarder,
                                    "wire sink dropped — discarding mixed packet"
                                );
                            }
                        }
                    }
                } else {
                    while let Some(item) = futures::StreamExt::next(&mut mix_rx).await {
                        if wire_sink.send(item).await.is_err() {
                            tracing::error!(
                                task = %HoprTransportProcess::MixerForwarder,
                                "wire sink dropped — discarding mixed packet"
                            );
                        }
                    }
                }

                tracing::warn!(
                    task = %HoprTransportProcess::MixerForwarder,
                    "long-running background task finished"
                );
            }),
        );

        // -- path cache background refresh (only when tokio runtime is available)
        #[cfg(feature = "runtime-tokio")]
        processes.insert(
            HoprTransportProcess::PathRefresh,
            hopr_utils::spawn_as_abortable!(self.path_planner.run_background_refresh()),
        );

        processes.insert(
            HoprTransportProcess::Medium,
            hopr_utils::spawn_as_abortable!(transport_layer_process().inspect(|_| tracing::warn!(
                task = %HoprTransportProcess::Medium,
                "long-running background task finished"
            ))),
        );

        let msg_protocol_bidirectional_channel_capacity =
            std::env::var("HOPR_INTERNAL_PROTOCOL_BIDIRECTIONAL_CHANNEL_CAPACITY")
                .ok()
                .and_then(|s| s.trim().parse::<usize>().ok())
                .filter(|&c| c > 0)
                .unwrap_or(16_384);

        debug!(
            capacity = msg_protocol_bidirectional_channel_capacity,
            "creating protocol bidirectional channel"
        );
        let (tx_from_protocol, rx_from_protocol) =
            bounded_sink_channel::<(HoprPseudonym, ApplicationDataIn)>(msg_protocol_bidirectional_channel_capacity);

        // === START === cover traffic control
        // Allocate a cover traffic tag from the session telemetry partition to avoid
        // collisions with session and probing tags.
        let cover_traffic_allocated_tag = self
            .session_telemetry_tag_allocator
            .allocate()
            .ok_or_else(|| HoprTransportError::Api("failed to allocate cover traffic tag".into()))?;
        let cover_traffic_tag: Tag = cover_traffic_allocated_tag.value().into();

        // filter out the known cover traffic not to lose processing time with it
        // The allocated tag is moved into the closure to keep it alive for the transport lifetime.
        let rx_from_protocol = rx_from_protocol.filter_map(move |(pseudonym, data)| {
            let _keep_alive = &cover_traffic_allocated_tag;
            async move { (data.data.application_tag != cover_traffic_tag).then_some((pseudonym, data)) }
        });

        // prepare a cover traffic stream
        let cover_traffic_stream = CoverTrafficGeneration::build(&cover_traffic).filter_map(move |routing| {
            let start =
                hopr_api::types::crypto_random::random_integer(0, Some((RANDOM_DATA.len() - 100) as u64)) as usize;
            let data = &RANDOM_DATA[start..start + 100];

            futures::future::ready(if let Ok(data) = ApplicationData::new(cover_traffic_tag, data) {
                Some((routing, ApplicationDataOut::with_no_packet_info(data)))
            } else {
                tracing::error!("failed to construct cover traffic packet");
                None
            })
        });

        // merge cover traffic with other outgoing data
        let merged_unresolved_output_data =
            select_with_strategy(unresolved_routing_msg_rx, cover_traffic_stream, |_: &mut ()| {
                futures::stream::PollNext::Left
            });

        // === END === cover traffic control

        // We have to resolve DestinationRouting -> ResolvedTransportRouting before
        // sending the external packets to the transport pipeline. Concurrency matches
        // the encoder stage (output_concurrency) to avoid head-of-line blocking on
        // cache-miss path lookups.
        let path_planner = self.path_planner.clone();
        let distress_threshold = self.cfg.packet.surb_store.distress_threshold;
        let routing_concurrency = {
            let avail = std::thread::available_parallelism()
                .ok()
                .map(|n| n.get())
                .unwrap_or(1)
                .max(1)
                * 8;
            self.cfg
                .packet
                .pipeline
                .output_concurrency
                .filter(|&n| n > 0)
                .unwrap_or(avail)
        };
        let all_resolved_external_msg_rx = crate::path::resolve::resolve_routing_stage(
            merged_unresolved_output_data,
            move |size_hint, max_surbs, unresolved| {
                let path_planner = path_planner.clone();
                async move { path_planner.resolve_routing(size_hint, max_surbs, unresolved).await }
            },
            distress_threshold,
            routing_concurrency,
            crate::path::resolve::surb_resolution_wait(self.cfg.packet.pipeline.surb_resolution_wait),
        );

        let channels_dst = self
            .chain_api
            .domain_separators()
            .await
            .map_err(HoprTransportError::chain)?
            .channel;

        // The SSA generator is dimensioned from the global PIX config (not per
        // session) because `handle_ssa_request` (SessionManager) validates that the
        // Exit's negotiated quota matches the session's `pix_ssa_quota` before any
        // client commitments are generated, and the Exit's `new_exit_commitment`
        // bounds-checks polys_per_ssa and shares_per_poly.  The session quota is
        // a subset of what the global generator covers, so one generator suffices.
        //
        // Validated here rather than left to the constructor: `PixGlobalConfig` carries more than
        // the three fields `SsaGeneratorConfig` covers, and this used to be a SAFETY comment
        // asserting that it had already been validated "before this code runs" via
        // `#[validate(nested)]` — but nothing in this crate calls `validate()`, so the guarantee
        // rested entirely on every caller remembering to.
        validator::Validate::validate(&self.cfg.pix)
            .map_err(|error| HoprTransportError::Api(format!("invalid PIX configuration: {error}")))?;
        // Checked rather than `as`-cast: the validation above already bounds all three, but a
        // truncating cast would turn a future widening of any of those ranges into a silently
        // wrong-but-valid-looking dimension rather than a startup error.
        fn narrow<T: TryFrom<usize>>(value: usize, field: &str) -> errors::Result<T> {
            T::try_from(value).map_err(|_| HoprTransportError::Api(format!("PIX {field} out of range: {value}")))
        }
        // `try_new`, not `new`: the dimensions come from operator configuration, so a range
        // violation is a startup error to report rather than a panic to take down the node.
        let ssa_generator = Arc::new(
            hopr_protocol_pix::SsaShareGenerator::<HoprPixSpec>::try_new(hopr_protocol_pix::SsaGeneratorConfig {
                polynomials_per_ssa: narrow(self.cfg.pix.num_ssa_parts, "num_ssa_parts")?,
                threshold: narrow(self.cfg.pix.ssa_part_size, "ssa_part_size")?,
                surplus_shares: narrow(self.cfg.pix.surplus_shares(), "additional_shares")?,
            })
            .map_err(|error| HoprTransportError::Api(format!("invalid SSA generator configuration: {error}")))?,
        );

        let surb_round_trips = protocol::surb_telemetry::SurbRoundTripRegistry::default();

        let pipeline_builder = HoprPacketPipelineBuilder::new()
            .identity((&self.chain_key, &self.packet_key))
            .transport((mixing_channel_tx, wire_msg_rx))
            .api((tx_from_protocol, all_resolved_external_msg_rx))
            .surb_store(self.path_planner.surb_store.clone())
            .chain_api(self.chain_api.clone())
            .ticket_factory(ticket_factory)
            .ssa_generator(ssa_generator.clone())
            .channels_dst(channels_dst)
            .with_counters(self.counters.clone())
            .with_surb_telemetry(
                surb_round_trips.clone(),
                protocol::surb_telemetry::path_slots_of(self.graph.clone()),
            )
            .with_config(self.cfg.packet);

        // ── PixToolbox for the SessionManager ────────────────────────────
        // The SessionManager needs a PixToolbox on all node types to handle
        // PIX protocol messages (SsaRequest on Entry, SsaCommit on Exit).
        // Only Exit nodes construct an SsaReconstructor and wire SSA recovery
        // events back through the packet pipeline. Relay nodes do NOT
        // participate in PIX share processing at the pipeline level.
        // Entry nodes get a bare-bones PixToolbox (share_generator + dummy
        // reconstructor) to handle SsaRequest, but do not use the
        // reconstructor for SSA recovery.
        let pix_toolbox = match (role, exit_ack_share) {
            (protocol::NodeType::Exit, Some(ref ssa_events)) => {
                let (pix_tools, pipeline_builder) = wire_exit_pix(
                    pipeline_builder,
                    ssa_generator.clone(),
                    ssa_reconstructor(&self.cfg.pix)?,
                    ssa_events,
                    self.smgr.clone(),
                    &mut processes,
                );

                let pipeline_processes = pipeline_builder.build_for_exit();
                processes.extend_from(pipeline_processes);
                Some(pix_tools)
            }
            (protocol::NodeType::Relay, None) => {
                // Pure relay nodes do not participate in PIX — do not create a
                // PixToolbox so the SessionManager rejects UsePIX
                // pre-emptively during session initiation.
                let pipeline_processes = pipeline_builder.with_ticket_events(ticket_events).build_for_relay();
                processes.extend_from(pipeline_processes);
                None
            }
            (protocol::NodeType::Relay, Some(ref ssa_events)) => {
                // A relay that also acts as an Exit (has exit_ack_share) needs
                // a full PixToolbox to handle the PIX handshake — the same wiring as the
                // `(Exit, Some)` arm above, differing only in the terminal below.
                let (pix_tools, pipeline_builder) = wire_exit_pix(
                    pipeline_builder,
                    ssa_generator.clone(),
                    ssa_reconstructor(&self.cfg.pix)?,
                    ssa_events,
                    self.smgr.clone(),
                    &mut processes,
                );

                let pipeline_processes = pipeline_builder.with_ticket_events(ticket_events).build_for_relay();
                processes.extend_from(pipeline_processes);
                Some(pix_tools)
            }
            (protocol::NodeType::Entry, Some(ref ssa_events)) => {
                // Entry nodes need a bare-bones PixToolbox (share_generator only)
                // to handle incoming SsaRequest messages from the Exit.
                // No SSA reconstruction needed on Entry — forward events to the
                // PIX event broadcast so subscribers (e.g. tests) see them.
                // Configured like any other, not left on the defaults: this reconstructor is called
                // "dummy" because Entry does not reconstruct, but it is not provably unreachable —
                // the comment below records that an inbound `UsePIX` does reach `handle_ssa_commit`
                // here. A knob that binds on some node roles and not others is the same defect
                // shape as a comment quoting a constant it does not reference.
                let dummy_reconstructor = ssa_reconstructor(&self.cfg.pix)?;
                let (pix_tools, session_pix_events) = PixToolbox::new(ssa_generator.clone(), dummy_reconstructor);
                processes.insert(
                    HoprTransportProcess::PixEvents,
                    hopr_utils::spawn_as_abortable!(
                        // The same mapping as the Exit and Relay arms, deliberately: `DepositNeeded`
                        // is not *expected* here, but it is reachable. `SessionManager` has no
                        // notion of node role and this one runs with a live toolbox — only the
                        // incoming-session notification is drained below — so an inbound
                        // `StartSession` carrying `UsePIX` is accepted and ends in
                        // `handle_ssa_commit` emitting it. Panicking would take the whole PIX event
                        // task with it, stranding this node's own outbound deposits.
                        session_pix_events
                            .map(session_pix_event_to_pix_event)
                            .map(Ok)
                            .forward(ssa_events.clone().sink_map_err(HoprTransportError::other))
                    ),
                );

                processes.extend_from(pipeline_builder.build_for_entry());
                Some(pix_tools)
            }
            (_, None) => {
                // Nodes without pix_events sink (no PIX configured at all)
                let pipeline_processes = match role {
                    protocol::NodeType::Relay => pipeline_builder.with_ticket_events(ticket_events).build_for_relay(),
                    protocol::NodeType::Exit => pipeline_builder.build_for_exit(),
                    protocol::NodeType::Entry => pipeline_builder.build_for_entry(),
                };
                processes.extend_from(pipeline_processes);
                None
            }
        };

        // ── Ssmgr startup ─────────────────────────────────────────────────
        // Entry nodes don't accept incoming sessions, relay/exit nodes do.
        let smgr_start_res = if role != protocol::NodeType::Entry {
            self.smgr.start(
                unresolved_routing_msg_tx.clone(),
                on_incoming_session.ok_or_else(|| {
                    HoprTransportError::Api("on_incoming_session channel is required for relay/exit nodes".into())
                })?,
                pix_toolbox,
            )
        } else {
            self.smgr
                .start(unresolved_routing_msg_tx.clone(), futures::sink::drain(), pix_toolbox)
        };

        smgr_start_res
            .map_err(|_| HoprTransportError::Api("failed to start session manager".into()))?
            .into_iter()
            .enumerate()
            .map(|(i, jh)| (HoprTransportProcess::SessionsManagement(i + 1), jh))
            .for_each(|(k, v)| {
                processes.insert(k, v);
            });
        // -- periodic counter flush
        let flush_counters = self.counters.clone();
        let flush_graph = self.graph.clone();
        let flush_me = *self.packet_key.public();
        let flush_interval = self.cfg.counter_flush_interval;
        processes.insert(
            HoprTransportProcess::CounterFlush,
            hopr_utils::spawn_as_abortable!(async move {
                use hopr_api::graph::traits::{EdgeObservableWrite, EdgeWeightType};

                futures_time::stream::interval(futures_time::time::Duration::from(flush_interval))
                    .for_each(|_| {
                        for (peer, num_packets, num_acks) in flush_counters.drain() {
                            tracing::trace!(
                                %peer,
                                num_packets,
                                num_acks,
                                "flushing protocol conformance counters"
                            );
                            flush_graph.upsert_edge(&flush_me, &peer, |obs| {
                                obs.record(EdgeWeightType::ImmediateProtocolConformance { num_packets, num_acks });
                            });
                        }
                        futures::future::ready(())
                    })
                    .await;
            }),
        );

        // -- periodic SURB round-trip flush
        tracing::info!(?role, "starting surb round-trip flush task");
        // Long enough to outlast the silence gate that produced the evidence, so a path that stays
        // dead is re-marked before the previous mark lapses, and short enough that a path which
        // recovers unnoticed returns to closed-loop control promptly.
        const RETURN_PATH_DEGRADED_GRACE: std::time::Duration = std::time::Duration::from_secs(10);
        let surb_flush_graph = self.graph.clone();
        let surb_flush_interval = self.cfg.surb_flush_interval;
        let surb_flush_smgr = self.smgr.clone();
        let surb_flush_chain = self.chain_api.clone();
        let surb_flush_planner = self.path_planner.clone();
        processes.insert(
            HoprTransportProcess::SurbFlush,
            hopr_utils::spawn_as_abortable!(async move {
                let mut episodes = protocol::return_path_recovery::ReturnPathEpisodes::new(RETURN_PATH_DEGRADED_GRACE);
                let mut ticks = futures_time::stream::interval(futures_time::time::Duration::from(surb_flush_interval));

                while ticks.next().await.is_some() {
                    // Detection before the drain: `degraded_destinations` reads the counts the
                    // flush is about to reset.
                    let silent = surb_round_trips.degraded_destinations();

                    // The graph has to see this interval's counts *before* anything re-plans on it,
                    // otherwise the re-plan that the silence just triggered reads a graph one whole
                    // tick behind the evidence that triggered it.
                    protocol::surb_telemetry::flush_into(
                        &surb_round_trips,
                        &surb_flush_graph,
                        hopr_utils::platform::time::native::current_time()
                            .as_unix_timestamp()
                            .as_millis(),
                    );

                    // Re-plan first, refill only if re-planning moved traffic. Run the other way
                    // round the refill mints SURBs onto the very route the planner is abandoning --
                    // see `protocol::return_path_recovery`.
                    //
                    // Borrowed rather than cloned per call: the callbacks are `FnMut`, so anything
                    // they capture has to survive being invoked once per silent destination.
                    let (planner, chain, smgr) = (&surb_flush_planner, &surb_flush_chain, &surb_flush_smgr);
                    for step in episodes
                        .tick(
                            silent,
                            |destination| async move { planner.recompute_paths_from(&destination).await },
                            |destination| async move {
                                // Sessions name their destination by its chain address, this
                                // telemetry by its packet key, and a `NodeId` holding one is never
                                // equal to a `NodeId` holding the other -- so the match has to be
                                // made on a resolved form, not on the enum.
                                //
                                // The counterparty has to still be receiving SURBs to reply with,
                                // and its silence has by now convinced our balancer that it is well
                                // stocked -- so tell the Sessions routed there to stop believing
                                // that estimate while the evidence says otherwise.
                                chain
                                    .packet_key_to_chain_key(&destination)
                                    .ok()
                                    .flatten()
                                    .map(hopr_api::types::internal::prelude::NodeId::Chain)
                                    .map(|n| smgr.mark_return_path_degraded(&n, RETURN_PATH_DEGRADED_GRACE))
                                    .unwrap_or(0)
                            },
                        )
                        .await
                    {
                        match step {
                            protocol::return_path_recovery::RecoveryStep::Replanned { destination, moved } => {
                                tracing::info!(%destination, entries = moved, "return path silent, re-planned")
                            }
                            protocol::return_path_recovery::RecoveryStep::Refilled { destination, sessions } => {
                                tracing::info!(%destination, sessions, "refilling behind the re-plan")
                            }
                        }
                    }
                }
            }),
        );

        // -- network probing
        let manual_ping_channel_capacity = std::env::var("HOPR_INTERNAL_MANUAL_PING_CHANNEL_CAPACITY")
            .ok()
            .and_then(|s| s.trim().parse::<usize>().ok())
            .filter(|&c| c > 0)
            .unwrap_or(128);
        debug!(capacity = manual_ping_channel_capacity, "Creating manual ping channel");
        let (manual_ping_tx, manual_ping_rx_raw) =
            crossfire::mpsc::bounded_async::<(OffchainPublicKey, PingQueryReplier)>(manual_ping_channel_capacity);
        let manual_ping_rx = manual_ping_rx_raw.into_stream();

        let probe = Probe::new(self.cfg.probe, self.probing_tag_allocator.clone());

        let (probing_processes, probe_classifier) = probe
            .continuously_scan(
                unresolved_routing_msg_tx.clone(),
                manual_ping_rx,
                cover_traffic,
                self.graph.clone(),
            )
            .await;

        processes.flat_map_extend_from(probing_processes, HoprTransportProcess::Probing);

        // manual ping
        self.ping
            .clone()
            .set(Pinger::new(
                PingConfig {
                    timeout: self.cfg.probe.timeout,
                },
                manual_ping_tx,
            ))
            .map_err(|_| HoprTransportError::Api("must set the ticket aggregation writer only once".into()))?;

        // Wire incoming: cover-traffic-filtered stream → probe classify → (session dispatch).
        // This stage must run in a background task, so the pipeline drains even when the
        // caller discards the returned HoprSocket (e.g. edge-node builder).
        //
        // The channel uses a resilient for_each rather than .forward() so that a disconnected
        // receiver (HoprSocket dropped without consuming) logs an error and continues rather
        // than collapsing the entire ingress pipeline. Callers should use HoprSocket::reader()
        // and actively drain the stream; see hopr-lib builder for the reference drain.
        let (on_incoming_data_tx, on_incoming_data_rx) =
            bounded_sink_channel::<ApplicationDataIn>(msg_protocol_bidirectional_channel_capacity);
        let smgr = self.smgr.clone();
        let unresolved_routing_msg_tx_for_task = unresolved_routing_msg_tx.clone();
        processes.insert(
            HoprTransportProcess::SessionsManagement(0),
            hopr_utils::spawn_as_abortable!(async move {
                probe_classifier
                    .filter_stream(unresolved_routing_msg_tx_for_task, rx_from_protocol)
                    .filter_map(move |(pseudonym, data)| {
                        hopr_transport_session::counters::DISPATCH_MESSAGE_CALLS
                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        // `dispatch_message` is synchronous and lock-free (moka::sync + crossfire
                        // try_send); it never blocks. However, the crossfire stream that feeds
                        // this loop does not participate in tokio's cooperative budget, so
                        // `future::ready` — which is always Poll::Ready — would let this task
                        // monopolize its worker thread under a saturated inbound queue (Processed
                        // items are filtered to None before the fold's tx.send().await, the only
                        // other suspension point on the hot path). Calling consume_budget() here
                        // integrates this loop into tokio's coop scheduler: it is a cheap
                        // thread-local decrement on the fast path and only actually yields
                        // (~every 128 polls) when the budget is exhausted, giving co-located
                        // tasks (including the SPHINX-crypto Rayon pipeline) a chance to run.
                        let result = match smgr.dispatch_message(pseudonym, data) {
                            Ok(DispatchResult::Processed) => {
                                tracing::trace!("message dispatch completed");
                                None
                            }
                            Ok(DispatchResult::Unrelated(data)) => {
                                tracing::trace!("unrelated message dispatch completed");
                                Some(data)
                            }
                            Err(error) => {
                                tracing::error!(%error, "error while dispatching packet in the session manager");
                                None
                            }
                        };
                        async move {
                            hopr_utils::runtime::prelude::consume_budget().await;
                            result
                        }
                    })
                    .fold(on_incoming_data_tx, |mut tx, data| async move {
                        if tx.send(data).await.is_err() {
                            tracing::error!(
                                task = %HoprTransportProcess::SessionsManagement(0),
                                "incoming-data channel disconnected — dropping unrelated packet; \
                                 HoprSocket must be consumed or drained by the caller"
                            );
                        }
                        tx
                    })
                    .await;
                tracing::warn!(
                    task = %HoprTransportProcess::SessionsManagement(0),
                    "long-running background task finished"
                );
            }),
        );

        // Populate the OnceLock at the end, making sure everything before didn't fail.
        self.network
            .clone()
            .set(transport_network)
            .map_err(|_| HoprTransportError::Api("transport network viewer already set".into()))?;

        Ok((
            (on_incoming_data_rx.boxed(), unresolved_routing_msg_tx).into(),
            processes,
        ))
    }

    #[tracing::instrument(level = "debug", skip(self))]
    pub async fn ping(
        &self,
        peer: &OffchainPublicKey,
    ) -> errors::Result<(std::time::Duration, <Graph as NetworkGraphView>::Observed)> {
        let me: &OffchainPublicKey = self.packet_key.public();
        if peer == me {
            return Err(HoprTransportError::Api("ping to self does not make sense".into()));
        }

        let pinger = self
            .ping
            .get()
            .ok_or_else(|| HoprTransportError::Api("ping processing is not yet initialized".into()))?;

        let latency = (*pinger).ping(peer).await?;

        if let Some(observations) = self.graph.edge(me, peer) {
            Ok((latency, observations))
        } else {
            Err(HoprTransportError::Api(format!(
                "no observations available for peer {peer}",
            )))
        }
    }

    #[tracing::instrument(level = "debug", skip(self))]
    pub async fn new_session(
        &self,
        destination: Address,
        target: SessionTarget,
        cfg: SessionClientConfig,
    ) -> errors::Result<(HoprSession, HoprSessionConfigurator)> {
        let session = self.smgr.new_session(destination, target, cfg).await?;
        let id = *session.id();
        Ok((
            session,
            HoprSessionConfigurator {
                id,
                smgr: Arc::downgrade(&self.smgr),
            },
        ))
    }

    #[tracing::instrument(level = "debug", skip(self))]
    pub async fn listening_multiaddresses(&self) -> Vec<Multiaddr> {
        self.network
            .get()
            .ok_or_else(|| HoprTransportError::Api("transport network is not yet initialized".into()))
            .map(|network| network.listening_as().into_iter().collect())
            .unwrap_or_default()
    }

    #[tracing::instrument(level = "debug", skip(self))]
    pub fn announceable_multiaddresses(&self) -> Vec<Multiaddr> {
        let mut mas = self
            .local_multiaddresses()
            .into_iter()
            .filter(|ma| {
                crate::multiaddrs::is_supported(ma)
                    && (self.cfg.transport.announce_local_addresses || is_public_address(ma))
            })
            .map(|ma| strip_p2p_protocol(&ma))
            .filter(|v| !v.is_empty())
            .collect::<Vec<_>>();

        mas.sort_by(|l, r| {
            let is_left_dns = crate::multiaddrs::is_dns(l);
            let is_right_dns = crate::multiaddrs::is_dns(r);

            if !(is_left_dns ^ is_right_dns) {
                std::cmp::Ordering::Equal
            } else if is_left_dns {
                std::cmp::Ordering::Less
            } else {
                std::cmp::Ordering::Greater
            }
        });

        mas
    }

    /// Returns a reference to the network graph.
    pub fn graph(&self) -> &Graph {
        &self.graph
    }

    /// Returns a reference to the SURB store.
    ///
    /// Exposed so that chain-level channel events can invalidate stored SURBs whose return path
    /// starts with a relayer this node can no longer pay.
    pub fn surb_store(&self) -> &MemorySurbStore {
        &self.path_planner.surb_store
    }

    #[tracing::instrument(level = "debug", skip(self))]
    pub fn local_multiaddresses(&self) -> Vec<Multiaddr> {
        self.network
            .get()
            .map(|network| network.listening_as().into_iter().collect())
            .unwrap_or_else(|| {
                tracing::error!("transport network is not yet initialized, cannot fetch announced multiaddresses");
                self.my_multiaddresses.clone()
            })
    }

    #[tracing::instrument(level = "debug", skip(self))]
    pub async fn network_observed_multiaddresses(&self, peer: &OffchainPublicKey) -> Vec<Multiaddr> {
        match self
            .network
            .get()
            .ok_or_else(|| HoprTransportError::Api("transport network is not yet initialized".into()))
        {
            Ok(network) => network
                .multiaddress_of(&peer.into())
                .unwrap_or_default()
                .into_iter()
                .collect(),
            Err(error) => {
                tracing::error!(%error, "failed to get observed multiaddresses");
                return vec![];
            }
        }
    }

    #[tracing::instrument(level = "debug", skip(self))]
    pub async fn network_health(&self) -> Health {
        self.network
            .get()
            .ok_or_else(|| HoprTransportError::Api("transport network is not yet initialized".into()))
            .map(|network| network.health())
            .unwrap_or(Health::Red)
    }

    pub async fn network_connected_peers(&self) -> errors::Result<Vec<OffchainPublicKey>> {
        Ok(futures::stream::iter(
            self.network
                .get()
                .ok_or_else(|| {
                    tracing::error!("transport network is not yet initialized");
                    HoprTransportError::Api("transport network is not yet initialized".into())
                })?
                .connected_peers(),
        )
        .filter_map(|peer_id| async move {
            match peer_id_to_public_key(&peer_id) {
                Ok(key) => Some(key),
                Err(error) => {
                    tracing::warn!(%peer_id, %error, "failed to convert PeerId to OffchainPublicKey");
                    None
                }
            }
        })
        .collect()
        .await)
    }

    #[tracing::instrument(level = "debug", skip(self))]
    pub fn network_peer_observations(&self, peer: &OffchainPublicKey) -> Option<<Graph as NetworkGraphView>::Observed> {
        self.graph.edge(self.packet_key.public(), peer)
    }

    /// Get connected peers with quality higher than some value.
    #[tracing::instrument(level = "debug", skip(self))]
    pub async fn all_network_peers(
        &self,
        minimum_score: f64,
    ) -> errors::Result<Vec<(OffchainPublicKey, <Graph as NetworkGraphView>::Observed)>> {
        let me = self.packet_key.public();
        Ok(self
            .network_connected_peers()
            .await?
            .into_iter()
            .filter_map(|peer| {
                let observation = self.graph.edge(me, &peer);
                if let Some(info) = observation {
                    // An unobserved edge has no score and cannot clear any threshold.
                    if info.score().is_some_and(|score| score >= minimum_score) {
                        Some((peer, info))
                    } else {
                        None
                    }
                } else {
                    None
                }
            })
            .collect::<Vec<_>>())
    }
}

/// Maps a fully recovered SSA into the downstream event that carries the recovered deposit
/// key to the withdrawal strategy.
pub(crate) fn recovered_ssa_to_pix_event(
    rec: &RecoveredSsa<SimplePseudonym, <HoprPixSpec as PixSpec>::AddressPrivateKey>,
) -> PixEvent {
    PixEvent::PrivateKeyRecovered(PixPrivateKeyRecovered {
        id: (*rec.ssa_id.pseudonym(), rec.ssa_id.ssa_index()),
        secret: PixDepositSecret(rec.ssa.secret().clone()),
    })
}

/// How many PIX share resolutions may be dispatched into the [`SessionManager`] at once.
///
/// A dispatch is not cheap: `SsaAlmostRecovered` / `SsaRecovered` reach `request_next_ssa`, which
/// acquires a per-session lock (with a 30 s timeout), generates an Exit commitment on the blocking
/// pool and sends an `SsaRequest` over the network. All sessions share this one stream, so
/// dispatching sequentially lets a single slow or stalled session hold up PIX progress — and
/// therefore the pipelined next-SSA request — for every other session on the node.
const PIX_EVENT_DISPATCH_CONCURRENCY: usize = 64;

/// Maps a [`HoprSessionOutPixEvent`] into the upper-layer [`PixEvent`] carrying the deposit
/// instruction for the funding strategy.
fn session_pix_event_to_pix_event(event: HoprSessionOutPixEvent) -> PixEvent {
    match event {
        HoprSessionOutPixEvent::ReadyToDeposit(AgreedSsaQuota {
            ssa_id,
            deposit_address,
            quota_per_ssa,
        }) => PixEvent::NewDepositAddress(PixNewDepositAddress {
            id: (*ssa_id.pseudonym(), ssa_id.ssa_index()),
            address: deposit_address.into(),
            quota: quota_per_ssa,
            // `None` rather than the handshake's `deposit_data`: that field is carried by the Start
            // protocol's `SsaRequest` (`hopr-protocol-start`) and is not surfaced through
            // `AgreedSsaQuota`, so there is nothing here to forward yet. Threading it through is a
            // change to the Session layer's event types, not to this mapping.
            additional_data: None,
        }),
        HoprSessionOutPixEvent::DepositNeeded(
            AgreedSsaQuota {
                ssa_id,
                deposit_address,
                quota_per_ssa,
            },
            notifier,
        ) => PixEvent::DepositAddressReceived(PixDepositAddressReceived {
            id: (*ssa_id.pseudonym(), ssa_id.ssa_index()),
            address: deposit_address.into(),
            quota: quota_per_ssa,
            // See `NewDepositAddress` above: the handshake's `deposit_data` does not reach
            // `AgreedSsaQuota`, so there is nothing to forward yet.
            additional_data: None,
            deposit_updated: Some(notifier),
        }),
    }
}

/// Notifies the [`SessionManager`] about a single PIX share resolution and returns the
/// upper-layer [`PixEvent`] it produces, if any.
async fn dispatch_share_resolution(smgr: Arc<HoprSessionManager>, resolution: HoprShareResolution) -> Option<PixEvent> {
    match resolution {
        ShareResolution::RecoveredSsa(ssa_recovery_event) => {
            if let Err(error) = smgr
                .dispatch_pix_event(HoprSessionInPixEvent::SsaRecovered(ssa_recovery_event.ssa_id))
                .await
            {
                tracing::error!(%error, "failed to dispatch SSA recovery PIX event to the SessionManager");
            }
            Some(recovered_ssa_to_pix_event(&ssa_recovery_event))
        }
        ShareResolution::AlmostRecoveredSsa(ssa_id) => {
            if let Err(error) = smgr
                .dispatch_pix_event(HoprSessionInPixEvent::SsaAlmostRecovered(ssa_id))
                .await
            {
                tracing::error!(%error, %ssa_id, "failed to dispatch early SSA recovery event to the SessionManager");
            }
            None
        }
        ShareResolution::InvalidShares {
            peer,
            ssa_id,
            observed_total,
        } => {
            tracing::error!(
                %peer, %ssa_id, observed_total,
                "first RP relayer sent acknowledgement indicating invalid PIX share from Entry"
            );
            if let Err(error) = smgr
                .dispatch_pix_event(HoprSessionInPixEvent::UnverifiableShare(ssa_id))
                .await
            {
                tracing::error!(%error, %ssa_id, "failed to dispatch invalid share PIX event to the SessionManager");
            }
            None
        }
        // Nothing consumes recovery progress yet — the Exit-side PIX supervisor is what will, and it
        // is the only thing that can act on a running total. Dropped here rather than suppressed at
        // the reconstructor so that the emission contract (and its tests) live with the producer.
        //
        // Cheap to discard: this branch awaits nothing, so the resolution channel is drained faster
        // than the acknowledgement path can fill it.
        ShareResolution::Progress(progress) => {
            tracing::trace!(
                ssa_id = %progress.ssa_id,
                useful_shares = progress.useful_shares,
                target = progress.target_useful_shares,
                recovered_polynomials = progress.recovered_polynomials,
                "pix recovery progress"
            );
            None
        }
    }
}

/// Builds the Exit-side SSA reconstructor from operator configuration.
///
/// The single place `PixReconstructorConfig` becomes an `SsaReconstructorConfig`, called once per
/// node role that needs a reconstructor. Both production sites used to take
/// `SsaReconstructorConfig::default()` instead, which left all seven of the Exit's dials — capacity,
/// lifetimes, the acknowledgement window — unreachable from a config file.
///
/// `try_new` rather than `new`, for the same reason the generator uses it at the one call site
/// above: these values now come from an operator, so an out-of-range one is a startup error to
/// report and not a panic to take the node down with. `run_inner` has already validated
/// `self.cfg.pix` — `reconstructor` is `#[validate(nested)]` — so in practice this reports what that
/// validation would have caught anyway; it stays fallible because nothing in the type system says
/// the two must be called in that order.
fn ssa_reconstructor(cfg: &PixGlobalConfig) -> errors::Result<Arc<hopr_protocol_pix::SsaReconstructor<HoprPixSpec>>> {
    hopr_protocol_pix::SsaReconstructor::<HoprPixSpec>::try_new(cfg.reconstructor.into())
        .map(Arc::new)
        .map_err(|error| HoprTransportError::Api(format!("invalid SSA reconstructor configuration: {error}")))
}

/// Wires the Exit-side PIX machinery: [`PixToolbox`], the recovered-share channel and the
/// [`HoprTransportProcess::PixEvents`] task, returning the toolbox and the pipeline builder with
/// share processing attached.
///
/// Shared by the `(Exit, Some)` and `(Relay, Some)` arms of `run_inner`, which differ only in the
/// terminal `build_for_*` call and the relay's extra `with_ticket_events` step. Both need the exact
/// same wiring, and keeping it in one place is not cosmetic: the previous two copies had already
/// drifted, with the comment on the reconstructor config fixed in the Exit copy only.
///
/// The reconstructor arrives built, like the generator, so that [`ssa_reconstructor`] stays the one
/// place a config turns into one and this stays infallible.
///
/// The caller keeps the terminal, so this returns the builder rather than the built processes.
/// Only the `exit_ack_proc` and `ssa_events` type parameters change; the rest pass through
/// untouched, which is why none of them carry bounds here.
#[allow(clippy::type_complexity)]
fn wire_exit_pix<WIn, WOut, Chain, S, TFact, G, AppOut, AppIn, TEvt, PixEvt>(
    pipeline_builder: HoprPacketPipelineBuilder<WIn, WOut, Chain, S, TFact, G, AppOut, AppIn, TEvt>,
    ssa_generator: Arc<hopr_protocol_pix::SsaShareGenerator<HoprPixSpec>>,
    ssa_reconstructor: Arc<hopr_protocol_pix::SsaReconstructor<HoprPixSpec>>,
    ssa_events: &PixEvt,
    smgr: Arc<HoprSessionManager>,
    processes: &mut AbortableList<HoprTransportProcess>,
) -> (
    PixToolbox,
    HoprPacketPipelineBuilder<
        WIn,
        WOut,
        Chain,
        S,
        TFact,
        G,
        AppOut,
        AppIn,
        TEvt,
        Arc<hopr_protocol_pix::SsaReconstructor<HoprPixSpec>>,
        CrossfireSink<HoprShareResolution>,
    >,
)
where
    PixEvt: futures::Sink<PixEvent> + Clone + Unpin + Send + 'static,
    PixEvt::Error: std::error::Error + Clone + Sync + Send + 'static,
{
    let (pix_tools, session_pix_events) = PixToolbox::new(ssa_generator, ssa_reconstructor.clone());
    let (ssa_share_resolution_events_tx, ssa_share_resolution_events_rx) = bounded_sink_channel(1024);
    processes.insert(
        HoprTransportProcess::PixEvents,
        hopr_utils::spawn_as_abortable!(
            pix_event_stream(session_pix_events, ssa_share_resolution_events_rx, smgr)
                .map(Ok)
                .forward(ssa_events.clone().sink_map_err(HoprTransportError::other))
        ),
    );

    (
        pix_tools,
        pipeline_builder.with_exit_ack_share_processing(ssa_reconstructor, ssa_share_resolution_events_tx),
    )
}

/// Builds the merged PIX event stream feeding the upper layer.
///
/// Combines the Session-originated deposit instructions with the share resolutions coming out of
/// the packet pipeline. Share resolutions are dispatched into the `SessionManager` concurrently
/// (see [`PIX_EVENT_DISPATCH_CONCURRENCY`]).
///
/// Concurrency does not require ordering guarantees here: `request_next_ssa` serializes on a
/// per-session lock and re-checks the SSA index under it, so of the events belonging to one cycle
/// exactly one advances the index and the rest are recognised as stale and become no-ops,
/// regardless of the order in which they arrive.
fn pix_event_stream(
    session_pix_events: impl futures::Stream<Item = HoprSessionOutPixEvent> + Send + 'static,
    ssa_share_resolutions: impl futures::Stream<Item = HoprShareResolution> + Send + 'static,
    smgr: Arc<HoprSessionManager>,
) -> impl futures::Stream<Item = PixEvent> + Send + 'static {
    session_pix_events.map(session_pix_event_to_pix_event).merge(
        ssa_share_resolutions
            .then_concurrent(
                move |resolution| dispatch_share_resolution(smgr.clone(), resolution),
                PIX_EVENT_DISPATCH_CONCURRENCY,
            )
            .filter_map(futures::future::ready),
    )
}

// ---------------------------------------------------------------------------
// NetworkView impl for HoprTransport — wraps OnceLock<Net> access
// ---------------------------------------------------------------------------

impl<Chain, Graph, Net> NetworkView for HoprTransport<Chain, Graph, Net>
where
    Net: NetworkView + Send + Sync + 'static,
{
    fn listening_as(&self) -> std::collections::HashSet<Multiaddr> {
        self.network.get().map(|n| n.listening_as()).unwrap_or_default()
    }

    fn multiaddress_of(&self, peer: &PeerId) -> Option<std::collections::HashSet<Multiaddr>> {
        self.network.get()?.multiaddress_of(peer)
    }

    fn discovered_peers(&self) -> std::collections::HashSet<PeerId> {
        self.network.get().map(|n| n.discovered_peers()).unwrap_or_default()
    }

    fn connected_peers(&self) -> std::collections::HashSet<PeerId> {
        self.network.get().map(|n| n.connected_peers()).unwrap_or_default()
    }

    fn is_connected(&self, peer: &PeerId) -> bool {
        self.network.get().map(|n| n.is_connected(peer)).unwrap_or(false)
    }

    fn health(&self) -> Health {
        self.network.get().map(|n| n.health()).unwrap_or(Health::Red)
    }

    fn subscribe_network_events(
        &self,
    ) -> impl futures::Stream<Item = hopr_api::network::NetworkEvent> + Send + 'static {
        match self.network.get() {
            Some(n) => futures::future::Either::Left(n.subscribe_network_events()),
            None => futures::future::Either::Right(futures::stream::empty()),
        }
    }
}

// ---------------------------------------------------------------------------
// TransportOperations impl for HoprTransport
// ---------------------------------------------------------------------------

#[async_trait::async_trait]
impl<Chain, Graph, Net> hopr_api::node::TransportOperations for HoprTransport<Chain, Graph, Net>
where
    Chain: ChainReadChannelOperations
        + ChainReadAccountOperations
        + hopr_api::chain::ChainWriteTicketOperations
        + ChainKeyOperations
        + hopr_api::chain::ChainReadTicketOperations
        + ChainValues
        + Clone
        + Send
        + Sync
        + 'static,
    Graph: NetworkGraphView<NodeId = OffchainPublicKey>
        + NetworkGraphUpdate
        + hopr_api::graph::NetworkGraphWrite<NodeId = OffchainPublicKey>
        + hopr_api::graph::NetworkGraphTraverse<NodeId = OffchainPublicKey>
        + Clone
        + Send
        + Sync
        + 'static,
    <Graph as NetworkGraphView>::Observed: EdgeObservableRead + Send,
    <Graph as hopr_api::graph::NetworkGraphTraverse>::Observed: EdgeObservableRead + Send + 'static,
    <Graph as hopr_api::graph::NetworkGraphWrite>::Observed: hopr_api::graph::traits::EdgeObservableWrite + Send,
    Net: NetworkView + NetworkStreamControl + Clone + Send + Sync + 'static,
{
    type Error = errors::HoprTransportError;
    type Observable = <Graph as NetworkGraphView>::Observed;

    async fn ping(&self, key: &OffchainPublicKey) -> Result<(Duration, Self::Observable), Self::Error> {
        self.ping(key).await
    }

    async fn observed_multiaddresses(&self, key: &OffchainPublicKey) -> Vec<Multiaddr> {
        self.network_observed_multiaddresses(key).await
    }
}

/// Maximum application-layer payload that fits in a single HOPR sphinx packet (bytes).
pub const PACKET_PAYLOAD_SIZE: usize = hopr_crypto_packet::prelude::HoprPacket::PAYLOAD_SIZE;

#[cfg(test)]
mod pix_recovery_event_tests {
    use hopr_api::{
        node::PixEvent,
        types::{
            crypto::{
                keypairs::{Keypair, OffchainKeypair},
                types::{HalfKey, SimplePseudonym},
            },
            crypto_random::{Randomizable, random_bytes},
            internal::prelude::VerifiedAcknowledgement,
        },
    };
    use hopr_crypto_packet::HoprPixSpec;
    use hopr_protocol_pix::{
        EntryShareGenerator, ExitAcknowledgementShareProcessor, PixSpec, SsaGeneratorConfig, SsaId, SsaIndex,
        SsaReconstructor, SsaReconstructorConfig, SsaShareGenerator, TaggedEncryptedPartialSsaShare,
    };

    use super::recovered_ssa_to_pix_event;

    #[test]
    fn recovered_ssa_maps_to_private_key_event_with_correct_secret_and_id() -> anyhow::Result<()> {
        let cfg = SsaGeneratorConfig {
            polynomials_per_ssa: 2,
            threshold: 2,
            surplus_shares: 0,
        };
        let generator = SsaShareGenerator::<HoprPixSpec>::new(cfg);
        let reconstructor = SsaReconstructor::<HoprPixSpec>::new(SsaReconstructorConfig {
            early_recovery_threshold: 1.0,
            ..Default::default()
        });

        let pseudonym = SimplePseudonym::random();
        let peer = OffchainKeypair::random();
        let ssa_id = SsaId::new(pseudonym, SsaIndex::MIN);

        let client = generator.new_ssa_commitment(&pseudonym, SsaIndex::MIN)?;
        let server_commitment = reconstructor.new_exit_commitment(ssa_id, 2, 2)?;
        let expected_addr = HoprPixSpec::group_to_deposit_address(client.ssa_commitment + server_commitment)
            .ok_or_else(|| anyhow::anyhow!("deposit address"))?;
        client.process_into_reconstructor(&reconstructor)?;

        let mut acks = Vec::new();
        while let Some((msg, share)) = {
            let msg = random_bytes::<20>();
            generator.next_share(&pseudonym, &msg).map(|v| v.map(|u| (msg, u)))
        }? {
            let ack = HalfKey::random();
            let enc = share.share.encrypt(&share.id, &ack)?;
            reconstructor.insert_encrypted_share(
                peer.public(),
                ack.to_challenge()?,
                TaggedEncryptedPartialSsaShare::new(pseudonym, &msg, enc)?,
            )?;
            acks.push(VerifiedAcknowledgement::new(ack, &peer).leak());
        }

        let resolutions = reconstructor.acknowledge_shares(*peer.public(), acks)?;
        let rec = resolutions
            .into_iter()
            .find_map(|r| r.try_as_recovered_ssa())
            .ok_or_else(|| anyhow::anyhow!("expected a RecoveredSsa resolution"))?;

        assert_eq!(<HoprPixSpec as PixSpec>::DepositAddress::from(&rec.ssa), expected_addr);

        let PixEvent::PrivateKeyRecovered(pk) = recovered_ssa_to_pix_event(&rec) else {
            anyhow::bail!("expected PrivateKeyRecovered");
        };

        assert_eq!(pk.id, (pseudonym, ssa_id.ssa_index()));
        assert_eq!(pk.secret.0.as_ref(), rec.ssa.secret().as_ref());

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{
        Arc,
        atomic::{AtomicU64, Ordering},
    };

    use futures::StreamExt;
    use hopr_utils::runtime::prelude::{consume_budget, spawn, yield_now};

    /// Verifies that the `filter_map` closure in `SessionsManagement(0)` does not monopolize
    /// the tokio worker thread under a continuously saturated inbound stream.
    ///
    /// ### Background
    ///
    /// The inbound dispatch loop (`transport/hopr/src/lib.rs`) drives:
    ///
    /// ```text
    /// rx_from_protocol (crossfire, no coop budget)
    ///   → filter_stream
    ///   → filter_map(sync dispatch_message + consume_budget().await)
    ///   → fold(tx.send().await)          ← only fires for Unrelated items
    /// ```
    ///
    /// On the `Processed` hot path (common case), items become `None` and are filtered
    /// out before the `fold`'s `tx.send().await`, so the only suspension point is the
    /// `consume_budget()` call inside `filter_map`.
    ///
    /// ### What this test proves
    ///
    /// On a `current_thread` runtime (single worker, fully cooperative scheduling) a tight
    /// stream loop that never returns `Poll::Pending` starves every other spawned task until
    /// the stream is drained.  By replacing `future::ready(result)` with an `async` block
    /// that calls `consume_budget().await` before returning, we give the scheduler a chance
    /// to preempt the loop every ≈128 polls.  The "canary" task — a plain `yield_now` loop —
    /// *must* make forward progress while the dispatch stream is draining for the fix to be
    /// considered effective.
    ///
    /// Reverting the `consume_budget().await` to `future::ready(result)` will make this
    /// test fail: the canary will never be scheduled and its counter will remain 0.
    #[tokio::test(flavor = "current_thread")]
    async fn sessions_management_dispatch_loop_yields_to_scheduler() {
        // N large enough to trigger ≥1 genuine yield (budget = 128 per reset).
        const N: usize = 1_000;

        // Canary: counts how many times it was scheduled during the dispatch run.
        let counter = Arc::new(AtomicU64::new(0));
        let counter_clone = counter.clone();
        let canary = spawn(async move {
            loop {
                counter_clone.fetch_add(1, Ordering::Relaxed);
                yield_now().await;
            }
        });

        // Dispatch loop — same shape as the fixed SessionsManagement(0) hot path:
        //   sync compute → consume_budget().await → None (Processed path)
        let unrelated_count = futures::stream::iter(0..N)
            .filter_map(|_item| {
                // Dispatch result computed synchronously (no clone); budget consumed async.
                let result: Option<()> = None; // every item is "Processed"
                async move {
                    consume_budget().await;
                    result
                }
            })
            .fold(0u64, |acc, _| async move { acc + 1 })
            .await;

        assert_eq!(
            unrelated_count, 0,
            "all Processed items should be filtered out by filter_map"
        );

        // The canary must have been scheduled at least once while the dispatch loop ran.
        // If consume_budget() is removed and future::ready() is used instead, the loop
        // never returns Poll::Pending on the current_thread executor and this assertion fails.
        let progress = counter.load(Ordering::Relaxed);
        assert!(
            progress > 0,
            "canary task made no forward progress during the {N}-item dispatch run (counter = {progress}); the \
             dispatch loop is monopolizing the executor thread — ensure filter_map returns `async move {{ \
             consume_budget().await; result }}` rather than `future::ready(result)`"
        );

        canary.abort();
        canary.await.ok();
    }
}