hopr_transport_session/
lib.rs

1//! [`HoprSession`] object providing the session functionality over the HOPR transport
2//!
3//! The session proxies the user interactions with the transport to hide the
4//! advanced interactions and functionality.
5//!
6//! The [`SessionManager`] allows for automatic management of sessions via the Start protocol.
7//!
8//! This crate implements [RFC-0007](https://github.com/hoprnet/rfc/tree/main/rfcs/RFC-0007-session-protocol).
9
10pub(crate) mod balancer;
11pub mod counters;
12pub mod errors;
13pub mod flow_control;
14mod manager;
15/// Exit-side PIX supervision: the lifecycle state machine, the egress gate, and the per-session
16/// worker that drives them.
17pub(crate) mod supervision;
18#[cfg(feature = "telemetry")]
19mod telemetry;
20mod types;
21mod utils;
22
23pub use balancer::{AtomicSurbFlowEstimator, BalancerStateValues, MIN_BALANCER_SAMPLING_INTERVAL, SurbBalancerConfig};
24use hopr_api::types::internal::routing::RoutingOptions;
25pub use hopr_protocol_session::{AcknowledgementMode, flow_control::FlowControlConfig};
26pub use hopr_utils::network_types::types::*;
27pub use manager::{
28    DEFAULT_MAX_SSAS_PER_SSA_REQUEST, DEFAULT_SSAS_PER_SSA_REQUEST, DEPOSIT_DATA_REQUEST_TIMEOUT, DispatchResult,
29    DropReason, IncomingSessionPixConfig, MAX_ASSUMED_SESSION_PACKET_RATE, MAX_OVERLAPPING_BATCHES, MAX_SSA_BATCH_SIZE,
30    MIN_SURB_BUFFER_DURATION, PixToolbox, SessionManager, SessionManagerConfig, cycle_budget_for,
31    max_cycle_budget_for_quota, validate_incoming_session_pix_config,
32};
33pub use supervision::{FillRate, PixFillConfig, SupervisorConfig, validate_pix_supervision};
34/// The supervisor state machine and its event/action vocabulary, for `benches/supervisor_bench.rs`,
35/// and the egress gate, for `benches/gate_bench.rs`.
36///
37/// Behind the same `benchmark` gate as
38/// [`SessionManager::pre_populate_session`](crate::SessionManager::pre_populate_session), and for
39/// the same reason: a criterion bench is a separate crate, so a path that is `pub(crate)` is a path
40/// it cannot measure. Nothing here is part of the crate's supported surface.
41#[cfg(any(feature = "benchmark", test))]
42pub use supervision::{
43    GateBlockReason, GateClosed, GateVerdict, ServiceGate, SessionPixAction, SessionPixEvent, SessionPixSupervisor,
44};
45#[cfg(any(test, feature = "testing"))]
46pub mod testing;
47pub use hopr_api::types::internal::routing::DestinationRouting;
48pub use hopr_protocol_app::prelude::{ApplicationDataIn, ApplicationDataOut};
49pub use hopr_protocol_pix::{InvalidPixParams, PixParams};
50#[cfg(feature = "telemetry")]
51pub use telemetry::{SessionAckMode, SessionLifecycleState};
52#[cfg(any(test, feature = "testing"))]
53pub use testing::{MsgSender as MockMsgSender, SendMsg, mock_packet_planning, msg_type, start_msg_match};
54pub use types::{
55    AdmissionToken, AgreedSsaQuota, ClosureReason, DEFAULT_PIX_POLYS_PER_SSA, DEFAULT_PIX_SHARES_PER_POLY,
56    DEFAULT_PIX_SSA_QUOTA, DEFAULT_PIX_SURPLUS_SHARES, HoprSession, HoprSessionCapabilities, HoprSessionConfig,
57    HoprSessionInPixEvent, HoprSessionOutPixEvent, HoprStartProtocol, IncomingSession, LOCAL_PIX_SUITE,
58    PIX_QUOTA_BYTES_PER_SHARE, ServiceId, SessionAdmissionReply, SessionAdmissionSink, SessionId, SessionTarget,
59    SurbAdmissionControl,
60};
61#[cfg(feature = "runtime-tokio")]
62pub use utils::{transfer_session, transfer_session_datagram};
63
64/// Number of bytes that can be sent in a single Session protocol payload.
65///
66/// In other words, this is the effective payload capacity of a single Session segment.
67pub const SESSION_MTU: usize =
68    hopr_protocol_session::session_socket_mtu::<{ hopr_protocol_app::v1::ApplicationData::PAYLOAD_SIZE }>();
69
70/// Size of the HOPR SURB in bytes.
71///
72/// This is the re-export of [`hopr_crypto_packet::HoprSurb::SIZE`].
73pub const SURB_SIZE: usize = hopr_crypto_packet::HoprSurb::SIZE;
74
75flagset::flags! {
76    /// Individual capabilities of a Session.
77    #[repr(u8)]
78    #[derive(PartialOrd, Ord, strum::EnumString, strum::Display, serde_repr::Serialize_repr, serde_repr::Deserialize_repr)]
79    pub enum Capability : u8 {
80        /// Frame segmentation.
81        Segmentation = 0b0000_1000,
82        /// Frame retransmission (ACK-based)
83        ///
84        /// Implies [`Segmentation`].
85        RetransmissionAck = 0b0000_1100,
86        /// Frame retransmission (NACK-based)
87        ///
88        /// Implies [`Segmentation`].
89        RetransmissionNack = 0b000_1010,
90        /// UDP-like behavior: disable packet buffering, and — on stateless (non-retransmitting)
91        /// sessions — preserve datagram boundaries by emitting each write as exactly one frame
92        /// (delivered to the peer as exactly one read), regardless of `frame_mtu`. Use this for
93        /// datagram-oriented targets (e.g. UDP/WireGuard) that must not have datagrams split or
94        /// coalesced. On reliable (retransmitting) sessions only the buffering behavior applies;
95        /// boundary preservation is stateless-only (the NACK missing-segment bitmap cannot address
96        /// the segments of an oversized datagram frame).
97        ///
98        /// Boundary preservation holds only up to the reader's buffer: a single read cannot return
99        /// more bytes than its buffer, so a datagram larger than the peer's read buffer is still
100        /// delivered in multiple reads. Callers must therefore size the read buffer for the largest
101        /// datagram they need preserved. In the UDP-forwarding path the datagram size is bounded at
102        /// ingress by the forwarding buffer (`HOPR_UDP_BUFFER_SIZE`), so it never exceeds it there;
103        /// a datagram-mode frame may otherwise be as large as `64 * SESSION_MTU`.
104        ///
105        /// Implies [`Segmentation`].
106        NoDelay = 0b0000_1001,
107        /// Disable SURB-based egress rate control.
108        ///
109        /// This applies only to the recipient of the Session (Exit).
110        ///
111        /// If not set, the lower half of additional data may contain information about the desired SURB buffer size.
112        NoRateControl = 0b0001_0000,
113        /// Indicates to the Session recipient (Exit) that this Session should use the PIX protocol.
114        ///
115        /// The Exit may still refuse the Session if the offered quota
116        /// (from this node's  [`SsaShareGenerator`](hopr_protocol_pix::SsaShareGenerator)) falls
117        /// outside its acceptable range, or (with `enforce_pix`) if this flag is *absent*.
118        UsePIX = 0b0010_0000,
119    }
120}
121
122/// Set of Session [capabilities](Capability).
123pub type Capabilities = flagset::FlagSet<Capability>;
124
125/// Configuration for the session.
126///
127/// Relevant primarily for the client, since the server is only
128/// a reactive component in regard to the session concept.
129#[derive(Debug, PartialEq, Clone, smart_default::SmartDefault)]
130pub struct SessionClientConfig {
131    /// The forward path options for the session.
132    #[default(RoutingOptions::Hops(hopr_api::types::primitive::bounded::BoundedSize::MIN))]
133    pub forward_path_options: RoutingOptions,
134    /// The return path options for the session.
135    #[default(RoutingOptions::Hops(hopr_api::types::primitive::bounded::BoundedSize::MIN))]
136    pub return_path_options: RoutingOptions,
137    /// Capabilities offered by the session.
138    #[default(_code = "Capability::Segmentation.into()")]
139    pub capabilities: Capabilities,
140    /// Optional pseudonym used for the session. Mostly useful for testing only.
141    #[default(None)]
142    pub pseudonym: Option<hopr_api::types::internal::protocol::HoprPseudonym>,
143    /// Enable automatic SURB management for the Session.
144    #[default(Some(SurbBalancerConfig::default()))]
145    pub surb_management: Option<SurbBalancerConfig>,
146    /// Sets the maximum number of SURBs sent with Session data packets (if they fit).
147    ///
148    /// This does not affect `KeepAlive` messages used with SURB balancing, as they will always
149    /// carry the maximum number of SURBs possible. Setting this to a higher number will put additional CPU
150    /// pressure on the local node as it will generate the given number of SURBs for each data packet.
151    ///
152    /// The default `1` follows the balancer's target: data packets carry no SURBs while the
153    /// counterparty is estimated to have a full buffer. Values greater than `1` bypass this gate;
154    /// use them only if the underlying traffic is highly asymmetric. Setting this to `0` leaves
155    /// all SURB production to the SURB balancer (if configured).
156    ///
157    /// The maximum number is naturally limited by the maximum payload size of a HOPR packet.
158    ///
159    /// Default is `1`.
160    #[default(1)]
161    pub max_surbs_per_data_packet: usize,
162    /// Opt-in client-side send-window flow control for this session.
163    ///
164    /// `None` (the default) leaves the session unpaced — today's behaviour. `Some(..)` enables the
165    /// adaptive AIMD window on the entry (sending) side; use [`FlowControlConfig::default`] for the
166    /// verified clean profile or [`FlowControlConfig::robust`] for the tail-tolerance bundle. This is
167    /// the client's explicit dial (only meaningful on a reliable / `RetransmissionAck` session).
168    #[default(None)]
169    pub flow_control: Option<FlowControlConfig>,
170    /// Abandon the frame due next once the sequence has advanced this far past it, instead of
171    /// holding everything already received for the whole frame timeout.
172    ///
173    /// Head-of-line bound for this session's incoming direction. `None` inherits the node's
174    /// setting, `Some(0)` disables it here, `Some(n)` sets it.
175    ///
176    /// Worth setting per session because the right value tracks reordering depth -- throughput x
177    /// latency spread / frame size -- which is a property of the traffic, not of the node: a bulk
178    /// data session and a control session on the same node differ by orders of magnitude.
179    ///
180    /// Has no effect on a session carrying a retransmission capability, where a missing frame can
181    /// still be recovered and waiting for it is productive.
182    pub max_frames_behind_gap: Option<usize>,
183}
184
185#[cfg(test)]
186mod tests {
187    use hopr_api::types::{crypto_random::Randomizable, internal::prelude::HoprPseudonym};
188    use hopr_crypto_packet::prelude::HoprPacket;
189    use hopr_protocol_app::v1::ApplicationData;
190    use hopr_protocol_session::session_socket_mtu;
191    use hopr_protocol_start::{
192        ErrorIdentifier, KeepAliveMessage, StartChallenge, StartErrorReason, StartErrorType, StartEstablished,
193        StartInitiation,
194    };
195
196    use super::*;
197    use crate::types::HoprStartProtocol;
198
199    #[test]
200    fn test_session_mtu() {
201        assert_eq!(SESSION_MTU, session_socket_mtu::<{ ApplicationData::PAYLOAD_SIZE }>());
202        assert_eq!(1452, SESSION_MTU); // Needs to be changed when the Session MTU cap changes
203    }
204
205    #[test]
206    fn hopr_start_protocol_messages_must_fit_within_hopr_packet() -> anyhow::Result<()> {
207        let msg = HoprStartProtocol::StartSession(StartInitiation {
208            challenge: StartChallenge::MAX,
209            target: SessionTarget::TcpStream(SealedHost::Plain(
210                "example-of-a-very-very-long-second-level-name.on-a-very-very-long-domain-name.info:65530".parse()?,
211            )),
212            capabilities: Capabilities::full().into(),
213            additional_data: (u64::MAX as u128) << u64::BITS,
214        });
215
216        assert!(
217            msg.encode()?.1.len() <= HoprPacket::PAYLOAD_SIZE,
218            "StartSession must fit within {}",
219            HoprPacket::PAYLOAD_SIZE
220        );
221
222        let msg = HoprStartProtocol::SessionEstablished(StartEstablished {
223            orig_challenge: StartChallenge::MAX,
224            session_id: HoprPseudonym::random(),
225        });
226
227        assert!(
228            msg.encode()?.1.len() <= HoprPacket::PAYLOAD_SIZE,
229            "SessionEstablished must fit within {}",
230            HoprPacket::PAYLOAD_SIZE
231        );
232
233        let msg = HoprStartProtocol::SessionError(StartErrorType {
234            identifier: ErrorIdentifier::Challenge(StartChallenge::MAX),
235            reason: StartErrorReason::NoSlotsAvailable,
236        });
237
238        assert!(
239            msg.encode()?.1.len() <= HoprPacket::PAYLOAD_SIZE,
240            "SessionError must fit within {}",
241            HoprPacket::PAYLOAD_SIZE
242        );
243
244        let msg = HoprStartProtocol::KeepAlive(KeepAliveMessage {
245            session_id: HoprPseudonym::random(),
246            flags: None.into(),
247            additional_data: 0xffffffff,
248        });
249        assert!(
250            msg.encode()?.1.len() <= HoprPacket::PAYLOAD_SIZE,
251            "KeepAlive must fit within {}",
252            HoprPacket::PAYLOAD_SIZE
253        );
254
255        Ok(())
256    }
257
258    #[test]
259    fn hopr_start_protocol_message_session_initiation_message_should_allow_for_at_least_one_surb() -> anyhow::Result<()>
260    {
261        let msg = HoprStartProtocol::StartSession(StartInitiation {
262            challenge: StartChallenge::MAX,
263            target: SessionTarget::TcpStream(SealedHost::Plain(
264                "example-of-a-very-very-long-second-level-name.on-a-very-very-long-domain-name.info:65530".parse()?,
265            )),
266            capabilities: Capabilities::full().into(),
267            additional_data: (u64::MAX as u128) << u64::BITS,
268        });
269        let len = msg.encode()?.1.len();
270        assert!(
271            HoprPacket::max_surbs_with_message(len) >= 1,
272            "Hopr StartSession message size ({}) must allow for at least 1 SURB in packet",
273            len
274        );
275
276        Ok(())
277    }
278
279    #[test]
280    fn hopr_start_protocol_message_keep_alive_message_should_allow_for_maximum_surbs() -> anyhow::Result<()> {
281        let keep_alive = KeepAliveMessage {
282            session_id: HoprPseudonym::random(),
283            flags: None.into(),
284            additional_data: 0xffffffff,
285        };
286        let capacity = keep_alive.max_surbs()?;
287        let data = ApplicationData::try_from(HoprStartProtocol::KeepAlive(keep_alive))?;
288
289        // Eight generation-tagged SURBs fit in an empty packet, but the keep-alive and its tag
290        // together need one byte more than the remaining space.
291        assert_eq!(capacity, 7);
292        assert_eq!(capacity, HoprPacket::max_surbs_with_message(data.total_len()));
293
294        Ok(())
295    }
296}
hopr_transport_session/lib.rs

hopr_transport_session/
lib.rs
