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

hopr_transport_session/
lib.rs
