hopr_protocol_session/socket/
state.rs

1use hopr_utils::network_types::crossfire_sink::CrossfireSink;
2
3use crate::{
4    errors::SessionError,
5    processing::types::FrameInspector,
6    protocol::{FrameAcknowledgements, FrameId, Segment, SegmentId, SegmentRequest, SeqIndicator, SessionMessage},
7};
8
9/// Components the `SessionSocket` exposes to a [`SocketState`].
10///
11/// This is the primary communication interface between the state and the socket.
12pub struct SocketComponents<const C: usize> {
13    /// Allows inspecting incomplete frames that are currently held by the socket.
14    ///
15    /// Some states might strictly require a frame inspector and may therefore
16    /// return an error in [`SocketState::run`] if not present.
17    pub inspector: Option<FrameInspector>,
18    /// Allows emitting control messages to the socket.
19    ///
20    /// It is a regular `SessionMessage` injected into the downstream.
21    /// Strict-capacity bounded channel — capacity never inflates regardless of clone count.
22    pub ctl_tx: CrossfireSink<SessionMessage<C>>,
23}
24
25/// Abstraction of the `SessionSocket` state.
26pub trait SocketState<const C: usize>: Send {
27    /// Gets ID of this Session.
28    fn session_id(&self) -> &str;
29
30    /// Starts the necessary processes inside the state.
31    /// Should be idempotent if called multiple times.
32    fn run(&mut self, components: SocketComponents<C>) -> Result<(), SessionError>;
33
34    /// Stops processes inside the state for the given direction.
35    fn stop(&mut self) -> Result<(), SessionError>;
36
37    /// Called when the Socket receives a new segment from Downstream.
38    /// When the error is returned, the incoming segment is not passed Upstream.
39    fn incoming_segment(&mut self, id: &SegmentId, ind: SeqIndicator) -> Result<(), SessionError>;
40
41    /// Called when [segment retransmission request](SegmentRequest) is received from Downstream.
42    fn incoming_retransmission_request(&mut self, request: SegmentRequest<C>) -> Result<(), SessionError>;
43
44    /// Called when an [acknowledgement of frames](FrameAcknowledgements) is received from Downstream.
45    fn incoming_acknowledged_frames(&mut self, ack: FrameAcknowledgements<C>) -> Result<(), SessionError>;
46
47    /// Called when a complete Frame has been finalized from segments received from Downstream.
48    fn frame_complete(&mut self, id: FrameId) -> Result<(), SessionError>;
49
50    /// Called when a complete Frame emitted to Upstream in-sequence.
51    fn frame_emitted(&mut self, id: FrameId) -> Result<(), SessionError>;
52
53    /// Called when a frame could not be completed from the segments received from Downstream.
54    fn frame_discarded(&mut self, id: FrameId) -> Result<(), SessionError>;
55
56    /// Called when a segment of a Frame was sent to the Downstream.
57    fn segment_sent(&mut self, segment: &Segment) -> Result<(), SessionError>;
58
59    /// Convenience method to dispatch a `SessionMessage` to one of the available handlers.
60    fn incoming_message(&mut self, message: &SessionMessage<C>) -> Result<(), SessionError> {
61        match &message {
62            SessionMessage::Segment(s) => self.incoming_segment(&s.id(), s.seq_flags),
63            SessionMessage::Request(r) => self.incoming_retransmission_request(r.clone()),
64            SessionMessage::Acknowledge(a) => self.incoming_acknowledged_frames(a.clone()),
65        }
66    }
67}
68
69/// Represents a stateless Session socket.
70///
71/// Does nothing by default, only logs warnings and events for tracing.
72#[derive(Clone)]
73pub struct Stateless<const C: usize>(String);
74
75impl<const C: usize> Stateless<C> {
76    pub(crate) fn new<I: std::fmt::Display>(session_id: I) -> Self {
77        Self(session_id.to_string())
78    }
79}
80
81impl<const C: usize> SocketState<C> for Stateless<C> {
82    fn session_id(&self) -> &str {
83        &self.0
84    }
85
86    fn run(&mut self, _: SocketComponents<C>) -> Result<(), SessionError> {
87        Ok(())
88    }
89
90    fn stop(&mut self) -> Result<(), SessionError> {
91        Ok(())
92    }
93
94    fn incoming_segment(&mut self, _: &SegmentId, _: SeqIndicator) -> Result<(), SessionError> {
95        Ok(())
96    }
97
98    fn incoming_retransmission_request(&mut self, _: SegmentRequest<C>) -> Result<(), SessionError> {
99        Ok(())
100    }
101
102    fn incoming_acknowledged_frames(&mut self, _: FrameAcknowledgements<C>) -> Result<(), SessionError> {
103        Ok(())
104    }
105
106    fn frame_complete(&mut self, _: FrameId) -> Result<(), SessionError> {
107        Ok(())
108    }
109
110    fn frame_emitted(&mut self, _: FrameId) -> Result<(), SessionError> {
111        Ok(())
112    }
113
114    fn frame_discarded(&mut self, _: FrameId) -> Result<(), SessionError> {
115        Ok(())
116    }
117
118    fn segment_sent(&mut self, _: &Segment) -> Result<(), SessionError> {
119        Ok(())
120    }
121}
122
123#[cfg(test)]
124mod tests {
125    use std::{collections::HashSet, time::Duration};
126
127    use anyhow::Context;
128    use futures::{AsyncReadExt, AsyncWriteExt};
129    use futures_time::future::FutureExt;
130
131    use super::*;
132    #[cfg(feature = "telemetry")]
133    use crate::socket::telemetry::NoopTracker;
134    use crate::{
135        SessionSocket, SessionSocketConfig,
136        utils::test::{FaultyNetworkConfig, setup_alice_bob},
137    };
138
139    const FRAME_SIZE: usize = 1500;
140
141    const MTU: usize = 1000;
142
143    mockall::mock! {
144        SockState {}
145        impl SocketState<MTU> for SockState {
146            fn session_id(&self) -> &str;
147            fn run(&mut self, components: SocketComponents<MTU>) -> Result<(), SessionError>;
148            fn stop(&mut self) -> Result<(), SessionError>;
149            fn incoming_segment(&mut self, id: &SegmentId, ind: SeqIndicator) -> Result<(), SessionError>;
150            fn incoming_retransmission_request(&mut self, request: SegmentRequest<MTU>) -> Result<(), SessionError>;
151            fn incoming_acknowledged_frames(&mut self, ack: FrameAcknowledgements<MTU>) -> Result<(), SessionError>;
152            fn frame_complete(&mut self, id: FrameId) -> Result<(), SessionError>;
153            fn frame_emitted(&mut self, id: FrameId) -> Result<(), SessionError>;
154            fn frame_discarded(&mut self, id: FrameId) -> Result<(), SessionError>;
155            fn segment_sent(&mut self, segment: &Segment) -> Result<(), SessionError>;
156        }
157    }
158
159    #[derive(Clone)]
160    struct CloneableMockState<'a>(std::sync::Arc<std::sync::Mutex<MockSockState>>, &'a str);
161
162    impl<'a> CloneableMockState<'a> {
163        pub fn new(state: MockSockState, id: &'a str) -> Self {
164            Self(std::sync::Arc::new(std::sync::Mutex::new(state)), id)
165        }
166    }
167
168    impl SocketState<MTU> for CloneableMockState<'_> {
169        fn session_id(&self) -> &str {
170            let _ = self.0.lock().unwrap().session_id();
171            self.1
172        }
173
174        fn run(&mut self, components: SocketComponents<MTU>) -> Result<(), SessionError> {
175            tracing::debug!(id = self.1, "run called");
176            self.0.lock().unwrap().run(components)
177        }
178
179        fn stop(&mut self) -> Result<(), SessionError> {
180            tracing::debug!(id = self.1, "stop called");
181            self.0.lock().unwrap().stop()
182        }
183
184        fn incoming_segment(&mut self, id: &SegmentId, ind: SeqIndicator) -> Result<(), SessionError> {
185            tracing::debug!(id = self.1, "incoming_segment called");
186            self.0.lock().unwrap().incoming_segment(id, ind)
187        }
188
189        fn incoming_retransmission_request(&mut self, request: SegmentRequest<MTU>) -> Result<(), SessionError> {
190            tracing::debug!(id = self.1, "incoming_retransmission_request called");
191            self.0.lock().unwrap().incoming_retransmission_request(request)
192        }
193
194        fn incoming_acknowledged_frames(&mut self, ack: FrameAcknowledgements<MTU>) -> Result<(), SessionError> {
195            tracing::debug!(id = self.1, "incoming_acknowledged_frames called");
196            self.0.lock().unwrap().incoming_acknowledged_frames(ack)
197        }
198
199        fn frame_complete(&mut self, id: FrameId) -> Result<(), SessionError> {
200            tracing::debug!(id = self.1, "frame_complete called");
201            self.0.lock().unwrap().frame_complete(id)
202        }
203
204        fn frame_emitted(&mut self, id: FrameId) -> Result<(), SessionError> {
205            tracing::debug!(id = self.1, "frame_received called");
206            self.0.lock().unwrap().frame_emitted(id)
207        }
208
209        fn frame_discarded(&mut self, id: FrameId) -> Result<(), SessionError> {
210            tracing::debug!(id = self.1, "frame_discarded called");
211            self.0.lock().unwrap().frame_discarded(id)
212        }
213
214        fn segment_sent(&mut self, segment: &Segment) -> Result<(), SessionError> {
215            tracing::debug!(id = self.1, "segment_sent called");
216            self.0.lock().unwrap().segment_sent(segment)
217        }
218    }
219
220    #[test_log::test(tokio::test)]
221    async fn session_socket_must_correctly_dispatch_segment_and_frame_state_events() -> anyhow::Result<()> {
222        const NUM_FRAMES: usize = 2;
223
224        const NUM_SEGMENTS: usize = NUM_FRAMES * FRAME_SIZE / MTU + 1;
225
226        let mut alice_seq = mockall::Sequence::new();
227        let mut alice_state = MockSockState::new();
228        alice_state.expect_session_id().return_const("alice".into());
229
230        alice_state
231            .expect_run()
232            .once()
233            .in_sequence(&mut alice_seq)
234            .return_once(|_| Ok::<_, SessionError>(()));
235        alice_state
236            .expect_segment_sent()
237            .times(NUM_SEGMENTS)
238            .in_sequence(&mut alice_seq)
239            .returning(|_| Ok::<_, SessionError>(()));
240        alice_state
241            .expect_stop()
242            .once()
243            .in_sequence(&mut alice_seq)
244            .return_once(|| Ok::<_, SessionError>(()));
245        alice_state
246            .expect_segment_sent() // terminating segment
247            .once()
248            .in_sequence(&mut alice_seq)
249            .return_once(|_| Ok::<_, SessionError>(()));
250        // PinnedDrop on SessionSocket calls state.stop() again on drop.
251        alice_state
252            .expect_stop()
253            .once()
254            .in_sequence(&mut alice_seq)
255            .return_once(|| Ok::<_, SessionError>(()));
256
257        let mut bob_seq = mockall::Sequence::new();
258        let mut bob_state = MockSockState::new();
259        bob_state.expect_session_id().return_const("bob".into());
260
261        bob_state
262            .expect_run()
263            .once()
264            .in_sequence(&mut bob_seq)
265            .return_once(|_| Ok::<_, SessionError>(()));
266        bob_state
267            .expect_incoming_segment()
268            .times(NUM_SEGMENTS - 1)
269            .in_sequence(&mut bob_seq)
270            .returning(|_, _| Ok::<_, SessionError>(()));
271        bob_state
272            .expect_frame_complete()
273            .once()
274            .in_sequence(&mut bob_seq)
275            .with(mockall::predicate::eq(2))
276            .returning(|_| Ok::<_, SessionError>(()));
277        bob_state
278            .expect_frame_discarded()
279            .once()
280            .in_sequence(&mut bob_seq)
281            .with(mockall::predicate::eq(1))
282            .returning(|_| Ok::<_, SessionError>(()));
283        bob_state
284            .expect_frame_emitted()
285            .once()
286            .in_sequence(&mut bob_seq)
287            .with(mockall::predicate::eq(2))
288            .returning(|_| Ok::<_, SessionError>(()));
289        bob_state
290            .expect_stop()
291            .once()
292            .in_sequence(&mut bob_seq)
293            .return_once(|| Ok::<_, SessionError>(()));
294        bob_state
295            .expect_segment_sent() // terminating segment
296            .once()
297            .in_sequence(&mut bob_seq)
298            .return_once(|_| Ok::<_, SessionError>(()));
299        // PinnedDrop on SessionSocket calls state.stop() again on drop.
300        bob_state
301            .expect_stop()
302            .once()
303            .in_sequence(&mut bob_seq)
304            .return_once(|| Ok::<_, SessionError>(()));
305
306        let (alice, bob) = setup_alice_bob::<MTU>(
307            FaultyNetworkConfig {
308                avg_delay: Duration::from_millis(10),
309                ids_to_drop: HashSet::from_iter([0_usize]),
310                ..Default::default()
311            },
312            None,
313            None,
314        );
315
316        let cfg = SessionSocketConfig {
317            frame_size: FRAME_SIZE,
318            frame_timeout: Duration::from_millis(55),
319            ..Default::default()
320        };
321
322        let mut alice_socket = SessionSocket::new(
323            alice,
324            CloneableMockState::new(alice_state, "alice"),
325            cfg,
326            #[cfg(feature = "telemetry")]
327            NoopTracker,
328        )?;
329        let mut bob_socket = SessionSocket::new(
330            bob,
331            CloneableMockState::new(bob_state, "bob"),
332            cfg,
333            #[cfg(feature = "telemetry")]
334            NoopTracker,
335        )?;
336
337        let alice_sent_data = hopr_types::crypto_random::random_bytes::<{ NUM_FRAMES * FRAME_SIZE }>();
338        alice_socket
339            .write_all(&alice_sent_data)
340            .timeout(futures_time::time::Duration::from_secs(2))
341            .await
342            .context("write_all timeout")??;
343        alice_socket.flush().await?;
344
345        // One entire frame is discarded
346        let mut bob_recv_data = [0u8; (NUM_FRAMES - 1) * FRAME_SIZE];
347        bob_socket
348            .read_exact(&mut bob_recv_data)
349            .timeout(futures_time::time::Duration::from_secs(2))
350            .await
351            .context("read_exact timeout")??;
352
353        tracing::debug!("stopping");
354        alice_socket.close().await?;
355        bob_socket.close().await?;
356
357        Ok(())
358    }
359}
hopr_protocol_session/socket/state.rs

hopr_protocol_session/socket/
state.rs
