hopr_protocol_session/socket/
mod.rs

1//! This module defines the socket-like interface for Session protocol.
2
3pub mod ack_state;
4pub mod state;
5
6/// Contains socket statistics types.
7#[cfg(feature = "telemetry")]
8pub mod telemetry;
9
10use std::{
11    pin::Pin,
12    sync::{Arc, atomic::AtomicU32},
13    task::{Context, Poll},
14    time::Duration,
15};
16
17use futures::{FutureExt, SinkExt, StreamExt, TryStreamExt, future, future::AbortHandle};
18use futures_concurrency::stream::Merge;
19use state::{SocketComponents, SocketState, Stateless};
20use tracing::{Instrument, instrument};
21#[cfg(feature = "telemetry")]
22use {
23    strum::IntoDiscriminant,
24    telemetry::{SessionMessageDiscriminants, SessionTelemetryTracker},
25};
26
27use crate::{
28    errors::SessionError,
29    processing::{ReassemblerExt, SegmenterExt, SequencerExt, types::FrameInspector},
30    protocol::{OrderedFrame, SegmentRequest, SeqIndicator, SessionCodec, SessionMessage},
31};
32
33/// Configuration object for [`SessionSocket`].
34#[derive(Debug, Copy, Clone, Eq, PartialEq, smart_default::SmartDefault)]
35pub struct SessionSocketConfig {
36    /// The maximum size of a frame on the read/write interface of the [`SessionSocket`].
37    ///
38    /// The size is always greater or equal to the MTU `C` of the underlying transport, and
39    /// less or equal to:
40    /// - (`C` - `SessionMessage::SEGMENT_OVERHEAD`) * (`SeqIndicator::MAX` + 1) for stateless sockets, or
41    /// - (`C` - `SessionMessage::SEGMENT_OVERHEAD`) * min(`SeqIndicator::MAX` + 1,
42    ///   `SegmentRequest::MAX_MISSING_SEGMENTS_PER_FRAME`) for stateful sockets
43    ///
44    /// Default is 1500 bytes.
45    #[default(1500)]
46    pub frame_size: usize,
47    /// The maximum time to wait for a frame to be fully received.
48    ///
49    /// Default is 800 ms.
50    #[default(Duration::from_millis(800))]
51    pub frame_timeout: Duration,
52    /// Maximum number of segments to buffer in the downstream transport.
53    /// If 0 is given, the transport is unbuffered.
54    ///
55    /// Default is 0.
56    #[default(0)]
57    pub max_buffered_segments: usize,
58    /// Capacity of the frame reconstructor, the maximum number of incomplete frames, before
59    /// they are dropped.
60    ///
61    /// Default is 8192.
62    #[default(8192)]
63    pub capacity: usize,
64    /// Flushes data written to the socket immediately to the underlying transport.
65    ///
66    /// Default is false.
67    #[default(false)]
68    pub flush_immediately: bool,
69    /// Capacity of the control channel, the maximum number of outstanding control messages.
70    ///
71    /// This option affects stateful sockets.
72    ///
73    /// Default is 2048.
74    #[default(2048)]
75    pub control_channel_capacity: usize,
76}
77
78enum WriteState {
79    WriteOnly,
80    Writing,
81    Flushing(usize),
82}
83
84/// Socket-like object implementing the Session protocol that can operate on any transport that
85/// implements [`futures::io::AsyncRead`] and [`futures::io::AsyncWrite`].
86///
87/// The [`SocketState`] `S` given during instantiation can facilitate reliable or unreliable
88/// behavior (see [`AcknowledgementState`](ack_state::AcknowledgementState))
89///
90/// The constant argument `C` specifies the MTU in bytes of the underlying transport.
91#[pin_project::pin_project(PinnedDrop)]
92pub struct SessionSocket<const C: usize, S: SocketState<C>> {
93    state: S,
94    // This is where upstream writes the to-be-segmented frame data to
95    upstream_frames_in: Pin<Box<dyn futures::io::AsyncWrite + Send>>,
96    // This is where upstream reads the reconstructed frame data from
97    downstream_frames_out: Pin<Box<dyn futures::io::AsyncRead + Send>>,
98    write_state: WriteState,
99}
100
101// Ensure `state.stop()` runs even if the socket is dropped without
102// `poll_close` (e.g. its owning task was aborted). Without this, the
103// detached tasks spawned in `SocketState::run` keep their own channel
104// senders and never terminate, pinning the FrameInspector, ring-buffer,
105// and ack buffers per leaked session.
106#[pin_project::pinned_drop]
107impl<const C: usize, S: SocketState<C>> PinnedDrop for SessionSocket<C, S> {
108    fn drop(self: Pin<&mut Self>) {
109        let this = self.project();
110        if let Err(error) = this.state.stop() {
111            tracing::debug!(%error, "state.stop on SessionSocket drop failed");
112        }
113    }
114}
115
116impl<const C: usize> SessionSocket<C, Stateless<C>> {
117    /// Creates a new stateless socket suitable for fast UDP-like communication.
118    ///
119    /// Note that this results in a faster socket than if created via [`SessionSocket::new`] with
120    /// [`Stateless`]. This is because the frame inspector does not need to be instantiated.
121    pub fn new_stateless<T, I>(
122        id: I,
123        transport: T,
124        cfg: SessionSocketConfig,
125        #[cfg(feature = "telemetry")] stats: impl SessionTelemetryTracker + Clone + Send + 'static,
126    ) -> Result<Self, SessionError>
127    where
128        T: futures::io::AsyncRead + futures::io::AsyncWrite + Send + Unpin + 'static,
129        I: std::fmt::Display + Clone,
130    {
131        // The maximum frame size in a stateless socket is only bounded by the size of the SeqIndicator
132        let frame_size = cfg.frame_size.clamp(
133            C,
134            (C - SessionMessage::<C>::SEGMENT_OVERHEAD) * (SeqIndicator::MAX + 1) as usize,
135        );
136
137        // Segment data incoming/outgoing using underlying transport
138        let mut framed = asynchronous_codec::Framed::new(transport, SessionCodec::<C>);
139
140        // Check if we allow sending multiple segments to downstream in a single write
141        // The HWM cannot be 0 bytes
142        framed.set_send_high_water_mark(1.max(cfg.max_buffered_segments * C));
143
144        // Downstream transport
145        let (packets_out, packets_in) = framed.split();
146
147        // If needed, add also stats to individual stages.
148        #[cfg(feature = "telemetry")]
149        let (s0, s1, s2, s3) = { (stats.clone(), stats.clone(), stats.clone(), stats.clone()) };
150
151        // Pipeline IN: Data incoming from Upstream
152        let upstream_frames_in = packets_out
153            .with(move |segment| {
154                #[cfg(feature = "telemetry")]
155                s0.outgoing_message(SessionMessageDiscriminants::Segment);
156
157                future::ok::<_, SessionError>(SessionMessage::<C>::Segment(segment))
158            })
159            .segmenter_with_terminating_segment::<C>(frame_size);
160
161        let last_emitted_frame = Arc::new(AtomicU32::new(0));
162        let last_emitted_frame_clone = last_emitted_frame.clone();
163
164        // Debug tracing spans for individual pipeline stages
165        let stage1_span = tracing::debug_span!("SessionSocket::packets_in::pre_reassembly", session_id = %id);
166        let stage2_span = tracing::debug_span!("SessionSocket::packets_in::pre_sequencing", session_id = %id);
167        let stage3_span = tracing::debug_span!("SessionSocket::packets_in::post_sequencing", session_id = %id);
168
169        let (packets_in_abort_handle, packets_in_abort_reg) = AbortHandle::new_pair();
170
171        // Pipeline OUT: Packets incoming from Downstream
172        // Continue receiving packets from downstream, unless we received a terminating frame.
173        // Once the terminating frame is received, the `packets_in_abort_handle` is triggered, terminating the pipeline.
174        let downstream_frames_out = futures::stream::Abortable::new(packets_in, packets_in_abort_reg)
175            // Filter-out segments that we've seen already
176            .filter_map(move |packet| {
177                let _span = stage1_span.enter();
178                futures::future::ready(match packet {
179                    Ok(packet) => {
180                        packet.try_as_segment().filter(|s| {
181                            #[cfg(feature = "telemetry")]
182                            s1.incoming_message(SessionMessageDiscriminants::Segment);
183
184                            // Filter old frame ids to save space in the Reassembler
185                            let last_emitted_id = last_emitted_frame.load(std::sync::atomic::Ordering::Relaxed);
186                            if s.frame_id <= last_emitted_id {
187                                tracing::warn!(frame_id = s.frame_id, last_emitted_id, "frame already seen");
188                                false
189                            } else {
190                                true
191                            }
192                        })
193                    }
194                    Err(error) => {
195                        tracing::error!(%error, "unparseable packet");
196                        #[cfg(feature = "telemetry")]
197                        s1.error();
198                        None
199                    }
200                })
201            })
202            // Reassemble the segments into frames
203            .reassembler(cfg.frame_timeout, cfg.capacity)
204            // Discard frames that we could not reassemble
205            .filter_map(move |maybe_frame| {
206                let _span = stage2_span.enter();
207                futures::future::ready(match maybe_frame {
208                    Ok(frame) => {
209                        #[cfg(feature = "telemetry")]
210                        s2.frame_completed();
211                        Some(OrderedFrame(frame))
212                    }
213                    Err(error) => {
214                        tracing::error!(%error, "failed to reassemble frame");
215                        #[cfg(feature = "telemetry")]
216                        s2.incomplete_frame();
217                        None
218                    }
219                })
220            })
221            // Put the frames into the correct sequence by Frame Ids
222            .sequencer(cfg.frame_timeout, cfg.capacity)
223            // Discard frames missing from the sequence
224            .filter_map(move |maybe_frame| {
225                let _span = stage3_span.enter();
226                future::ready(match maybe_frame {
227                    Ok(frame) => {
228                        last_emitted_frame_clone.store(frame.0.frame_id, std::sync::atomic::Ordering::Relaxed);
229                        if frame.0.is_terminating {
230                            tracing::warn!("terminating frame received");
231                            packets_in_abort_handle.abort();
232                        }
233                        #[cfg(feature = "telemetry")]
234                        s3.frame_emitted();
235                        Some(Ok(frame.0))
236                    }
237                    // Downstream skips discarded frames
238                    Err(SessionError::FrameDiscarded(frame_id)) | Err(SessionError::IncompleteFrame(frame_id)) => {
239                        tracing::error!(frame_id, "frame discarded");
240                        #[cfg(feature = "telemetry")]
241                        s3.frame_discarded();
242                        None
243                    }
244                    Err(err) => {
245                        #[cfg(feature = "telemetry")]
246                        s3.error();
247                        Some(Err(std::io::Error::other(err)))
248                    }
249                })
250            })
251            .into_async_read();
252
253        Ok(Self {
254            state: Stateless::new(id),
255            upstream_frames_in: Box::pin(upstream_frames_in),
256            downstream_frames_out: Box::pin(downstream_frames_out),
257            write_state: if cfg.flush_immediately {
258                WriteState::Writing
259            } else {
260                WriteState::WriteOnly
261            },
262        })
263    }
264}
265
266impl<const C: usize, S: SocketState<C> + Clone + 'static> SessionSocket<C, S> {
267    /// Creates a stateful socket with frame inspection capabilities - suitable for communication
268    /// requiring TCP-like delivery guarantees.
269    pub fn new<T>(
270        transport: T,
271        mut state: S,
272        cfg: SessionSocketConfig,
273        #[cfg(feature = "telemetry")] stats: impl SessionTelemetryTracker + Clone + Send + 'static,
274    ) -> Result<Self, SessionError>
275    where
276        T: futures::io::AsyncRead + futures::io::AsyncWrite + Send + Unpin + 'static,
277    {
278        // The maximum frame size is reduced due to the size of the missing segment bitmap in SegmentRequests
279        let frame_size = cfg.frame_size.clamp(
280            C,
281            (C - SessionMessage::<C>::SEGMENT_OVERHEAD)
282                * SegmentRequest::<C>::MAX_MISSING_SEGMENTS_PER_FRAME.min((SeqIndicator::MAX + 1) as usize),
283        );
284
285        // Segment data incoming/outgoing using underlying transport
286        let mut framed = asynchronous_codec::Framed::new(transport, SessionCodec::<C>);
287
288        // Check if we allow sending multiple segments to downstream in a single write
289        // The HWM cannot be 0 bytes
290        framed.set_send_high_water_mark(1.max(cfg.max_buffered_segments * C));
291
292        // If needed, add also stats to individual stages.
293        #[cfg(feature = "telemetry")]
294        let (s0, s1, s2, s3) = { (stats.clone(), stats.clone(), stats.clone(), stats.clone()) };
295
296        // Downstream transport
297        let (packets_out, packets_in) = framed.split();
298
299        let inspector = FrameInspector::new(cfg.capacity);
300
301        tracing::debug!(
302            capacity = cfg.control_channel_capacity,
303            "creating session control channel"
304        );
305        let (ctl_tx, ctl_rx) = hopr_utils::network_types::crossfire_sink::bounded_sink_channel::<SessionMessage<C>>(
306            cfg.control_channel_capacity.max(128),
307        );
308        state.run(SocketComponents {
309            inspector: Some(inspector.clone()),
310            ctl_tx,
311        })?;
312
313        // Pipeline IN: Data incoming from Upstream
314        let (segments_tx, segments_rx) = futures::channel::mpsc::channel(cfg.capacity);
315        let mut st_1 = state.clone();
316        let upstream_frames_in = segments_tx
317            .with(move |segment| {
318                let _span =
319                    tracing::debug_span!("SessionSocket::packets_out::segmenter", session_id = st_1.session_id())
320                        .entered();
321                // The segment_sent event is raised only for segments coming from Upstream,
322                // not for the segments from the Control stream (= segment resends).
323                if let Err(error) = st_1.segment_sent(&segment) {
324                    tracing::debug!(session_id = st_1.session_id(), %error, "outgoing segment state update failed");
325                }
326                future::ok::<_, futures::channel::mpsc::SendError>(SessionMessage::<C>::Segment(segment))
327            })
328            .segmenter_with_terminating_segment::<C>(frame_size);
329
330        // We have to merge the streams here and spawn a special task for it
331        // Since the control messages from the State can come independent of Upstream writes.
332        hopr_utils::runtime::prelude::spawn(
333            (ctl_rx, segments_rx)
334                .merge()
335                .map(move |msg| {
336                    #[cfg(feature = "telemetry")]
337                    s0.outgoing_message(msg.discriminant());
338                    Ok(msg)
339                })
340                .forward(packets_out)
341                .map(move |result| match result {
342                    Ok(_) => tracing::debug!("outgoing packet processing done"),
343                    Err(error) => {
344                        tracing::error!(%error, "error while processing outgoing packets")
345                    }
346                })
347                .instrument(tracing::debug_span!(
348                    "SessionSocket::packets_out",
349                    session_id = state.session_id()
350                )),
351        );
352
353        let last_emitted_frame = Arc::new(AtomicU32::new(0));
354        let last_emitted_frame_clone = last_emitted_frame.clone();
355
356        let (packets_in_abort_handle, packets_in_abort_reg) = AbortHandle::new_pair();
357
358        // Pipeline OUT: Packets incoming from Downstream
359        let mut st_1 = state.clone();
360        let mut st_2 = state.clone();
361        let mut st_3 = state.clone();
362
363        // Continue receiving packets from downstream, unless we received a terminating frame.
364        // Once the terminating frame is received, the `packets_in_abort_handle` is triggered, terminating the pipeline.
365        let downstream_frames_out = futures::stream::Abortable::new(packets_in, packets_in_abort_reg)
366            // Filter out Session control messages and update the State, pass only Segments onwards
367            .filter_map(move |packet| {
368                let _span = tracing::debug_span!(
369                    "SessionSocket::packets_in::pre_reassembly",
370                    session_id = st_1.session_id()
371                )
372                .entered();
373                futures::future::ready(match packet {
374                    Ok(packet) => {
375                        if let Err(error) = st_1.incoming_message(&packet) {
376                            tracing::debug!(%error, "incoming message state update failed");
377                        }
378                        #[cfg(feature = "telemetry")]
379                        s1.incoming_message(packet.discriminant());
380
381                        // Filter old frame ids to save space in the Reassembler
382                        packet.try_as_segment().filter(|s| {
383                            let last_emitted_id = last_emitted_frame.load(std::sync::atomic::Ordering::Relaxed);
384                            if s.frame_id <= last_emitted_id {
385                                tracing::warn!(frame_id = s.frame_id, last_emitted_id, "frame already seen");
386                                false
387                            } else {
388                                true
389                            }
390                        })
391                    }
392                    Err(error) => {
393                        tracing::error!(%error, "unparseable packet");
394                        #[cfg(feature = "telemetry")]
395                        s1.error();
396                        None
397                    }
398                })
399            })
400            // Reassemble segments into frames
401            .reassembler_with_inspector(cfg.frame_timeout, cfg.capacity, inspector)
402            // Notify State once a frame has been reassembled, discard frames that we could not reassemble
403            .filter_map(move |maybe_frame| {
404                let _span = tracing::debug_span!(
405                    "SessionSocket::packets_in::pre_sequencing",
406                    session_id = st_2.session_id()
407                )
408                .entered();
409                futures::future::ready(match maybe_frame {
410                    Ok(frame) => {
411                        if let Err(error) = st_2.frame_complete(frame.frame_id) {
412                            tracing::error!(%error, "frame complete state update failed");
413                        }
414                        #[cfg(feature = "telemetry")]
415                        s2.frame_completed();
416                        Some(OrderedFrame(frame))
417                    }
418                    Err(error) => {
419                        tracing::error!(%error, "failed to reassemble frame");
420                        #[cfg(feature = "telemetry")]
421                        s2.incomplete_frame();
422                        None
423                    }
424                })
425            })
426            // Put the frames into the correct sequence by Frame Ids
427            .sequencer(cfg.frame_timeout, cfg.capacity)
428            // Discard frames missing from the sequence and
429            // notify the State about emitted or discarded frames
430            .filter_map(move |maybe_frame| {
431                let _span = tracing::debug_span!(
432                    "SessionSocket::packets_in::post_sequencing",
433                    session_id = st_3.session_id()
434                )
435                .entered();
436                // Filter out discarded Frames and dispatch events to the State if needed
437                future::ready(match maybe_frame {
438                    Ok(frame) => {
439                        if let Err(error) = st_3.frame_emitted(frame.0.frame_id) {
440                            tracing::error!(%error, "frame received state update failed");
441                        }
442                        last_emitted_frame_clone.store(frame.0.frame_id, std::sync::atomic::Ordering::Relaxed);
443                        if frame.0.is_terminating {
444                            tracing::warn!("terminating frame received");
445                            packets_in_abort_handle.abort();
446                        }
447                        #[cfg(feature = "telemetry")]
448                        s3.frame_emitted();
449                        Some(Ok(frame.0))
450                    }
451                    Err(SessionError::FrameDiscarded(frame_id)) | Err(SessionError::IncompleteFrame(frame_id)) => {
452                        if let Err(error) = st_3.frame_discarded(frame_id) {
453                            tracing::error!(%error, "frame discarded state update failed");
454                        }
455                        #[cfg(feature = "telemetry")]
456                        s3.frame_discarded();
457                        None // Downstream skips discarded frames
458                    }
459                    Err(err) => {
460                        #[cfg(feature = "telemetry")]
461                        s3.error();
462                        Some(Err(std::io::Error::other(err)))
463                    }
464                })
465            })
466            .into_async_read();
467
468        Ok(Self {
469            state,
470            upstream_frames_in: Box::pin(upstream_frames_in),
471            downstream_frames_out: Box::pin(downstream_frames_out),
472            write_state: if cfg.flush_immediately {
473                WriteState::Writing
474            } else {
475                WriteState::WriteOnly
476            },
477        })
478    }
479}
480
481impl<const C: usize, S: SocketState<C> + Clone + 'static> futures::io::AsyncRead for SessionSocket<C, S> {
482    #[instrument(name = "SessionSocket::poll_read", level = "trace", skip(self, cx, buf), fields(session_id = self.state.session_id(), len = buf.len()))]
483    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<std::io::Result<usize>> {
484        self.project().downstream_frames_out.as_mut().poll_read(cx, buf)
485    }
486}
487
488impl<const C: usize, S: SocketState<C> + Clone + 'static> futures::io::AsyncWrite for SessionSocket<C, S> {
489    #[instrument(name = "SessionSocket::poll_write", level = "trace", skip(self, cx, buf), fields(session_id = self.state.session_id(), len = buf.len()))]
490    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
491        let this = self.project();
492        loop {
493            match this.write_state {
494                WriteState::WriteOnly => {
495                    return this.upstream_frames_in.as_mut().poll_write(cx, buf);
496                }
497                WriteState::Writing => {
498                    let len = futures::ready!(this.upstream_frames_in.as_mut().poll_write(cx, buf))?;
499                    *this.write_state = WriteState::Flushing(len);
500                }
501                WriteState::Flushing(len) => {
502                    let res = futures::ready!(this.upstream_frames_in.as_mut().poll_flush(cx)).map(|_| *len);
503                    *this.write_state = WriteState::Writing;
504                    return Poll::Ready(res);
505                }
506            }
507        }
508    }
509
510    #[instrument(name = "SessionSocket::poll_flush", level = "trace", skip(self, cx), fields(session_id = self.state.session_id()))]
511    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
512        self.project().upstream_frames_in.as_mut().poll_flush(cx)
513    }
514
515    #[instrument(name = "SessionSocket::poll_close", level = "trace", skip(self, cx), fields(session_id = self.state.session_id()))]
516    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
517        let this = self.project();
518        let _ = this.state.stop();
519        this.upstream_frames_in.as_mut().poll_close(cx)
520    }
521}
522
523#[cfg(feature = "runtime-tokio")]
524impl<const C: usize, S: SocketState<C> + Clone + 'static> tokio::io::AsyncRead for SessionSocket<C, S> {
525    #[instrument(name = "SessionSocket::poll_read", level = "trace", skip(self, cx, buf), fields(session_id = self.state.session_id()))]
526    fn poll_read(
527        mut self: Pin<&mut Self>,
528        cx: &mut Context<'_>,
529        buf: &mut tokio::io::ReadBuf<'_>,
530    ) -> Poll<std::io::Result<()>> {
531        let slice = buf.initialize_unfilled();
532        let n = std::task::ready!(futures::AsyncRead::poll_read(self.as_mut(), cx, slice))?;
533        buf.advance(n);
534        Poll::Ready(Ok(()))
535    }
536}
537
538#[cfg(feature = "runtime-tokio")]
539impl<const C: usize, S: SocketState<C> + Clone + 'static> tokio::io::AsyncWrite for SessionSocket<C, S> {
540    #[instrument(name = "SessionSocket::poll_write", level = "trace", skip(self, cx, buf), fields(session_id = self.state.session_id(), len = buf.len()))]
541    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize, std::io::Error>> {
542        futures::AsyncWrite::poll_write(self.as_mut(), cx, buf)
543    }
544
545    #[instrument(name = "SessionSocket::poll_flush", level = "trace", skip(self, cx), fields(session_id = self.state.session_id()))]
546    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
547        futures::AsyncWrite::poll_flush(self.as_mut(), cx)
548    }
549
550    #[instrument(name = "SessionSocket::poll_shutdown", level = "trace", skip(self, cx), fields(session_id = self.state.session_id()))]
551    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
552        futures::AsyncWrite::poll_close(self.as_mut(), cx)
553    }
554}
555
556#[cfg(test)]
557mod tests {
558    use std::collections::HashSet;
559
560    use futures::{AsyncReadExt, AsyncWriteExt};
561    use futures_time::future::FutureExt;
562    use hopr_crypto_packet::prelude::HoprPacket;
563
564    use super::*;
565    #[cfg(feature = "telemetry")]
566    use crate::socket::telemetry::{NoopTracker, tests::TestTelemetryTracker};
567    use crate::{AcknowledgementState, AcknowledgementStateConfig, utils::test::*};
568
569    const MTU: usize = HoprPacket::PAYLOAD_SIZE;
570
571    const FRAME_SIZE: usize = 1500;
572
573    const DATA_SIZE: usize = 17 * MTU + 271; // Use some size not directly divisible by the MTU
574
575    #[test_log::test(tokio::test)]
576    async fn stateless_socket_unidirectional_should_work() -> anyhow::Result<()> {
577        let (alice, bob) = setup_alice_bob::<MTU>(FaultyNetworkConfig::default(), None, None);
578
579        let sock_cfg = SessionSocketConfig {
580            frame_size: FRAME_SIZE,
581            ..Default::default()
582        };
583
584        #[cfg(feature = "telemetry")]
585        let (alice_tracker, bob_tracker) = (TestTelemetryTracker::default(), TestTelemetryTracker::default());
586
587        let mut alice_socket = SessionSocket::<MTU, _>::new_stateless(
588            "alice",
589            alice,
590            sock_cfg,
591            #[cfg(feature = "telemetry")]
592            alice_tracker.clone(),
593        )?;
594        let mut bob_socket = SessionSocket::<MTU, _>::new_stateless(
595            "bob",
596            bob,
597            sock_cfg,
598            #[cfg(feature = "telemetry")]
599            bob_tracker.clone(),
600        )?;
601
602        let data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
603
604        alice_socket
605            .write_all(&data)
606            .timeout(futures_time::time::Duration::from_secs(2))
607            .await??;
608        alice_socket.flush().await?;
609
610        let mut bob_data = [0u8; DATA_SIZE];
611        bob_socket
612            .read_exact(&mut bob_data)
613            .timeout(futures_time::time::Duration::from_secs(2))
614            .await??;
615        assert_eq!(data, bob_data);
616
617        alice_socket.close().await?;
618        bob_socket.close().await?;
619
620        #[cfg(feature = "telemetry")]
621        {
622            insta::assert_yaml_snapshot!(alice_tracker);
623            insta::assert_yaml_snapshot!(bob_tracker);
624        }
625
626        Ok(())
627    }
628
629    #[test_log::test(tokio::test)]
630    async fn stateful_socket_unidirectional_should_work() -> anyhow::Result<()> {
631        let (alice, bob) = setup_alice_bob::<MTU>(FaultyNetworkConfig::default(), None, None);
632
633        let sock_cfg = SessionSocketConfig {
634            frame_size: FRAME_SIZE,
635            ..Default::default()
636        };
637
638        let ack_cfg = AcknowledgementStateConfig {
639            expected_packet_latency: Duration::from_millis(2),
640            acknowledgement_delay: Duration::from_millis(5),
641            ..Default::default()
642        };
643
644        let mut alice_socket = SessionSocket::<MTU, _>::new(
645            alice,
646            AcknowledgementState::new("alice", ack_cfg),
647            sock_cfg,
648            #[cfg(feature = "telemetry")]
649            NoopTracker,
650        )?;
651        let mut bob_socket = SessionSocket::<MTU, _>::new(
652            bob,
653            AcknowledgementState::new("bob", ack_cfg),
654            sock_cfg,
655            #[cfg(feature = "telemetry")]
656            NoopTracker,
657        )?;
658
659        let data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
660
661        alice_socket
662            .write_all(&data)
663            .timeout(futures_time::time::Duration::from_secs(2))
664            .await??;
665        alice_socket.flush().await?;
666
667        let mut bob_data = [0u8; DATA_SIZE];
668        bob_socket
669            .read_exact(&mut bob_data)
670            .timeout(futures_time::time::Duration::from_secs(2))
671            .await??;
672        assert_eq!(data, bob_data);
673
674        alice_socket.close().await?;
675        bob_socket.close().await?;
676
677        Ok(())
678    }
679
680    #[test_log::test(tokio::test)]
681    async fn stateless_socket_bidirectional_should_work() -> anyhow::Result<()> {
682        let (alice, bob) = setup_alice_bob::<MTU>(FaultyNetworkConfig::default(), None, None);
683
684        let sock_cfg = SessionSocketConfig {
685            frame_size: FRAME_SIZE,
686            ..Default::default()
687        };
688
689        #[cfg(feature = "telemetry")]
690        let (alice_tracker, bob_tracker) = (TestTelemetryTracker::default(), TestTelemetryTracker::default());
691
692        let mut alice_socket = SessionSocket::<MTU, _>::new_stateless(
693            "alice",
694            alice,
695            sock_cfg,
696            #[cfg(feature = "telemetry")]
697            alice_tracker.clone(),
698        )?;
699        let mut bob_socket = SessionSocket::<MTU, _>::new_stateless(
700            "bob",
701            bob,
702            sock_cfg,
703            #[cfg(feature = "telemetry")]
704            bob_tracker.clone(),
705        )?;
706
707        let alice_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
708        alice_socket
709            .write_all(&alice_sent_data)
710            .timeout(futures_time::time::Duration::from_secs(2))
711            .await??;
712        alice_socket.flush().await?;
713
714        let bob_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
715        bob_socket
716            .write_all(&bob_sent_data)
717            .timeout(futures_time::time::Duration::from_secs(2))
718            .await??;
719        bob_socket.flush().await?;
720
721        let mut bob_recv_data = [0u8; DATA_SIZE];
722        bob_socket
723            .read_exact(&mut bob_recv_data)
724            .timeout(futures_time::time::Duration::from_secs(2))
725            .await??;
726        assert_eq!(alice_sent_data, bob_recv_data);
727
728        let mut alice_recv_data = [0u8; DATA_SIZE];
729        alice_socket
730            .read_exact(&mut alice_recv_data)
731            .timeout(futures_time::time::Duration::from_secs(2))
732            .await??;
733        assert_eq!(bob_sent_data, alice_recv_data);
734
735        #[cfg(feature = "telemetry")]
736        {
737            insta::assert_yaml_snapshot!(alice_tracker);
738            insta::assert_yaml_snapshot!(bob_tracker);
739        }
740
741        Ok(())
742    }
743
744    #[test_log::test(tokio::test)]
745    async fn stateful_socket_bidirectional_should_work() -> anyhow::Result<()> {
746        let (alice, bob) = setup_alice_bob::<MTU>(FaultyNetworkConfig::default(), None, None);
747
748        // use hopr_network_types::capture::PcapIoExt;
749        // let (alice, bob) = (alice.capture("alice.pcap"), bob.capture("bob.pcap"));
750
751        let sock_cfg = SessionSocketConfig {
752            frame_size: FRAME_SIZE,
753            ..Default::default()
754        };
755
756        let ack_cfg = AcknowledgementStateConfig {
757            expected_packet_latency: Duration::from_millis(2),
758            acknowledgement_delay: Duration::from_millis(10),
759            ..Default::default()
760        };
761
762        let mut alice_socket = SessionSocket::<MTU, _>::new(
763            alice,
764            AcknowledgementState::new("alice", ack_cfg),
765            sock_cfg,
766            #[cfg(feature = "telemetry")]
767            NoopTracker,
768        )?;
769        let mut bob_socket = SessionSocket::<MTU, _>::new(
770            bob,
771            AcknowledgementState::new("bob", ack_cfg),
772            sock_cfg,
773            #[cfg(feature = "telemetry")]
774            NoopTracker,
775        )?;
776
777        let alice_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
778        alice_socket
779            .write_all(&alice_sent_data)
780            .timeout(futures_time::time::Duration::from_secs(2))
781            .await??;
782        alice_socket.flush().await?;
783
784        let bob_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
785        bob_socket
786            .write_all(&bob_sent_data)
787            .timeout(futures_time::time::Duration::from_secs(2))
788            .await??;
789        bob_socket.flush().await?;
790
791        let mut bob_recv_data = [0u8; DATA_SIZE];
792        bob_socket
793            .read_exact(&mut bob_recv_data)
794            .timeout(futures_time::time::Duration::from_secs(2))
795            .await??;
796        assert_eq!(alice_sent_data, bob_recv_data);
797
798        let mut alice_recv_data = [0u8; DATA_SIZE];
799        alice_socket
800            .read_exact(&mut alice_recv_data)
801            .timeout(futures_time::time::Duration::from_secs(2))
802            .await??;
803        assert_eq!(bob_sent_data, alice_recv_data);
804
805        Ok(())
806    }
807
808    #[test_log::test(tokio::test)]
809    async fn stateless_socket_unidirectional_should_work_with_mixing() -> anyhow::Result<()> {
810        let network_cfg = FaultyNetworkConfig {
811            mixing_factor: 10,
812            ..Default::default()
813        };
814
815        let (alice, bob) = setup_alice_bob::<MTU>(network_cfg, None, None);
816
817        let sock_cfg = SessionSocketConfig {
818            frame_size: FRAME_SIZE,
819            ..Default::default()
820        };
821
822        let mut alice_socket = SessionSocket::<MTU, _>::new_stateless(
823            "alice",
824            alice,
825            sock_cfg,
826            #[cfg(feature = "telemetry")]
827            NoopTracker,
828        )?;
829        let mut bob_socket = SessionSocket::<MTU, _>::new_stateless(
830            "bob",
831            bob,
832            sock_cfg,
833            #[cfg(feature = "telemetry")]
834            NoopTracker,
835        )?;
836
837        let data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
838        alice_socket
839            .write_all(&data)
840            .timeout(futures_time::time::Duration::from_secs(2))
841            .await??;
842        alice_socket.flush().await?;
843
844        let mut bob_recv_data = [0u8; DATA_SIZE];
845        bob_socket
846            .read_exact(&mut bob_recv_data)
847            .timeout(futures_time::time::Duration::from_secs(2))
848            .await??;
849        assert_eq!(data, bob_recv_data);
850
851        alice_socket.close().await?;
852        bob_socket.close().await?;
853
854        Ok(())
855    }
856
857    #[test_log::test(tokio::test)]
858    async fn stateful_socket_unidirectional_should_work_with_mixing() -> anyhow::Result<()> {
859        let network_cfg = FaultyNetworkConfig {
860            mixing_factor: 10,
861            ..Default::default()
862        };
863
864        let (alice, bob) = setup_alice_bob::<MTU>(network_cfg, None, None);
865
866        let sock_cfg = SessionSocketConfig {
867            frame_size: FRAME_SIZE,
868            ..Default::default()
869        };
870
871        let ack_cfg = AcknowledgementStateConfig {
872            expected_packet_latency: Duration::from_millis(2),
873            acknowledgement_delay: Duration::from_millis(5),
874            ..Default::default()
875        };
876
877        let mut alice_socket = SessionSocket::<MTU, _>::new(
878            alice,
879            AcknowledgementState::new("alice", ack_cfg),
880            sock_cfg,
881            #[cfg(feature = "telemetry")]
882            NoopTracker,
883        )?;
884        let mut bob_socket = SessionSocket::<MTU, _>::new(
885            bob,
886            AcknowledgementState::new("bob", ack_cfg),
887            sock_cfg,
888            #[cfg(feature = "telemetry")]
889            NoopTracker,
890        )?;
891
892        let data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
893        alice_socket
894            .write_all(&data)
895            .timeout(futures_time::time::Duration::from_secs(2))
896            .await??;
897        alice_socket.flush().await?;
898
899        let mut bob_recv_data = [0u8; DATA_SIZE];
900        bob_socket
901            .read_exact(&mut bob_recv_data)
902            .timeout(futures_time::time::Duration::from_secs(2))
903            .await??;
904        assert_eq!(data, bob_recv_data);
905
906        alice_socket.close().await?;
907        bob_socket.close().await?;
908
909        Ok(())
910    }
911
912    #[test_log::test(tokio::test)]
913    async fn stateless_socket_bidirectional_should_work_with_mixing() -> anyhow::Result<()> {
914        let network_cfg = FaultyNetworkConfig {
915            mixing_factor: 10,
916            ..Default::default()
917        };
918
919        let (alice, bob) = setup_alice_bob::<MTU>(network_cfg, None, None);
920
921        let sock_cfg = SessionSocketConfig {
922            frame_size: FRAME_SIZE,
923            ..Default::default()
924        };
925
926        let mut alice_socket = SessionSocket::<MTU, _>::new_stateless(
927            "alice",
928            alice,
929            sock_cfg,
930            #[cfg(feature = "telemetry")]
931            NoopTracker,
932        )?;
933        let mut bob_socket = SessionSocket::<MTU, _>::new_stateless(
934            "bob",
935            bob,
936            sock_cfg,
937            #[cfg(feature = "telemetry")]
938            NoopTracker,
939        )?;
940
941        let alice_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
942        alice_socket
943            .write_all(&alice_sent_data)
944            .timeout(futures_time::time::Duration::from_secs(2))
945            .await??;
946        alice_socket.flush().await?;
947
948        let bob_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
949        bob_socket
950            .write_all(&bob_sent_data)
951            .timeout(futures_time::time::Duration::from_secs(2))
952            .await??;
953        bob_socket.flush().await?;
954
955        let mut bob_recv_data = [0u8; DATA_SIZE];
956        bob_socket
957            .read_exact(&mut bob_recv_data)
958            .timeout(futures_time::time::Duration::from_secs(2))
959            .await??;
960        assert_eq!(alice_sent_data, bob_recv_data);
961
962        let mut alice_recv_data = [0u8; DATA_SIZE];
963        alice_socket
964            .read_exact(&mut alice_recv_data)
965            .timeout(futures_time::time::Duration::from_secs(2))
966            .await??;
967        assert_eq!(bob_sent_data, alice_recv_data);
968
969        alice_socket.close().await?;
970        bob_socket.close().await?;
971
972        Ok(())
973    }
974
975    #[test_log::test(tokio::test)]
976    async fn stateful_socket_bidirectional_should_work_with_mixing() -> anyhow::Result<()> {
977        let network_cfg = FaultyNetworkConfig {
978            mixing_factor: 10,
979            ..Default::default()
980        };
981
982        let (alice, bob) = setup_alice_bob::<MTU>(network_cfg, None, None);
983
984        let sock_cfg = SessionSocketConfig {
985            frame_size: FRAME_SIZE,
986            ..Default::default()
987        };
988
989        let ack_cfg = AcknowledgementStateConfig {
990            expected_packet_latency: Duration::from_millis(2),
991            acknowledgement_delay: Duration::from_millis(5),
992            ..Default::default()
993        };
994
995        let mut alice_socket = SessionSocket::<MTU, _>::new(
996            alice,
997            AcknowledgementState::new("alice", ack_cfg),
998            sock_cfg,
999            #[cfg(feature = "telemetry")]
1000            NoopTracker,
1001        )?;
1002        let mut bob_socket = SessionSocket::<MTU, _>::new(
1003            bob,
1004            AcknowledgementState::new("bob", ack_cfg),
1005            sock_cfg,
1006            #[cfg(feature = "telemetry")]
1007            NoopTracker,
1008        )?;
1009
1010        let alice_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
1011        alice_socket
1012            .write_all(&alice_sent_data)
1013            .timeout(futures_time::time::Duration::from_secs(2))
1014            .await??;
1015        alice_socket.flush().await?;
1016
1017        let bob_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
1018        bob_socket
1019            .write_all(&bob_sent_data)
1020            .timeout(futures_time::time::Duration::from_secs(2))
1021            .await??;
1022        bob_socket.flush().await?;
1023
1024        let mut bob_recv_data = [0u8; DATA_SIZE];
1025        bob_socket
1026            .read_exact(&mut bob_recv_data)
1027            .timeout(futures_time::time::Duration::from_secs(2))
1028            .await??;
1029        assert_eq!(alice_sent_data, bob_recv_data);
1030
1031        let mut alice_recv_data = [0u8; DATA_SIZE];
1032        alice_socket
1033            .read_exact(&mut alice_recv_data)
1034            .timeout(futures_time::time::Duration::from_secs(2))
1035            .await??;
1036        assert_eq!(bob_sent_data, alice_recv_data);
1037
1038        alice_socket.close().await?;
1039        bob_socket.close().await?;
1040
1041        Ok(())
1042    }
1043
1044    #[test_log::test(tokio::test)]
1045    async fn stateless_socket_unidirectional_should_should_skip_missing_frames() -> anyhow::Result<()> {
1046        let (alice, bob) = setup_alice_bob::<MTU>(
1047            FaultyNetworkConfig {
1048                avg_delay: Duration::from_millis(10),
1049                ids_to_drop: HashSet::from_iter([0_usize]),
1050                ..Default::default()
1051            },
1052            None,
1053            None,
1054        );
1055
1056        let alice_cfg = SessionSocketConfig {
1057            frame_size: FRAME_SIZE,
1058            ..Default::default()
1059        };
1060
1061        let bob_cfg = SessionSocketConfig {
1062            frame_size: FRAME_SIZE,
1063            frame_timeout: Duration::from_millis(55),
1064            ..Default::default()
1065        };
1066
1067        #[cfg(feature = "telemetry")]
1068        let (alice_tracker, bob_tracker) = (TestTelemetryTracker::default(), TestTelemetryTracker::default());
1069
1070        let mut alice_socket = SessionSocket::<MTU, _>::new_stateless(
1071            "alice",
1072            alice,
1073            alice_cfg,
1074            #[cfg(feature = "telemetry")]
1075            alice_tracker.clone(),
1076        )?;
1077        let mut bob_socket = SessionSocket::<MTU, _>::new_stateless(
1078            "bob",
1079            bob,
1080            bob_cfg,
1081            #[cfg(feature = "telemetry")]
1082            bob_tracker.clone(),
1083        )?;
1084
1085        let data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
1086        alice_socket
1087            .write_all(&data)
1088            .timeout(futures_time::time::Duration::from_secs(2))
1089            .await??;
1090        alice_socket.flush().await?;
1091        alice_socket.close().await?;
1092
1093        let mut bob_data = Vec::with_capacity(DATA_SIZE);
1094        bob_socket
1095            .read_to_end(&mut bob_data)
1096            .timeout(futures_time::time::Duration::from_secs(2))
1097            .await??;
1098
1099        // The whole first frame is discarded due to the missing first segment
1100        assert_eq!(data.len() - 1500, bob_data.len());
1101        assert_eq!(&data[1500..], &bob_data);
1102
1103        bob_socket.close().await?;
1104
1105        #[cfg(feature = "telemetry")]
1106        {
1107            insta::assert_yaml_snapshot!(alice_tracker);
1108            insta::assert_yaml_snapshot!(bob_tracker);
1109        }
1110
1111        Ok(())
1112    }
1113
1114    #[test_log::test(tokio::test)]
1115    async fn stateful_socket_unidirectional_should_should_not_skip_missing_frames() -> anyhow::Result<()> {
1116        let (alice, bob) = setup_alice_bob::<MTU>(
1117            FaultyNetworkConfig {
1118                avg_delay: Duration::from_millis(10),
1119                ids_to_drop: HashSet::from_iter([0_usize]),
1120                ..Default::default()
1121            },
1122            None,
1123            None,
1124        );
1125
1126        let alice_cfg = SessionSocketConfig {
1127            frame_size: FRAME_SIZE,
1128            ..Default::default()
1129        };
1130
1131        let bob_cfg = SessionSocketConfig {
1132            frame_size: FRAME_SIZE,
1133            frame_timeout: Duration::from_millis(1000),
1134            ..Default::default()
1135        };
1136
1137        let ack_cfg = AcknowledgementStateConfig {
1138            expected_packet_latency: Duration::from_millis(10),
1139            acknowledgement_delay: Duration::from_millis(40),
1140            ..Default::default()
1141        };
1142
1143        let mut alice_socket = SessionSocket::<MTU, _>::new(
1144            alice,
1145            AcknowledgementState::new("alice", ack_cfg),
1146            alice_cfg,
1147            #[cfg(feature = "telemetry")]
1148            NoopTracker,
1149        )?;
1150        let mut bob_socket = SessionSocket::<MTU, _>::new(
1151            bob,
1152            AcknowledgementState::new("bob", ack_cfg),
1153            bob_cfg,
1154            #[cfg(feature = "telemetry")]
1155            NoopTracker,
1156        )?;
1157
1158        let data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
1159
1160        let alice_jh = tokio::spawn(async move {
1161            alice_socket
1162                .write_all(&data)
1163                .timeout(futures_time::time::Duration::from_secs(5))
1164                .await??;
1165
1166            alice_socket.flush().await?;
1167
1168            // Alice has to keep reading so that it is ready for retransmitting
1169            let mut vec = Vec::new();
1170            alice_socket.read_to_end(&mut vec).await?;
1171            alice_socket.close().await?;
1172
1173            Ok::<_, std::io::Error>(vec)
1174        });
1175
1176        let mut bob_data = [0u8; DATA_SIZE];
1177        bob_socket
1178            .read_exact(&mut bob_data)
1179            .timeout(futures_time::time::Duration::from_secs(5))
1180            .await??;
1181        assert_eq!(data, bob_data);
1182
1183        bob_socket.close().await?;
1184
1185        let alice_recv = alice_jh.await??;
1186        assert!(alice_recv.is_empty());
1187
1188        Ok(())
1189    }
1190
1191    #[test_log::test(tokio::test)]
1192    async fn stateless_socket_bidirectional_should_should_skip_missing_frames() -> anyhow::Result<()> {
1193        let (alice, bob) = setup_alice_bob::<MTU>(
1194            FaultyNetworkConfig {
1195                avg_delay: Duration::from_millis(10),
1196                ids_to_drop: HashSet::from_iter([0_usize]),
1197                ..Default::default()
1198            },
1199            None,
1200            None,
1201        );
1202
1203        let alice_cfg = SessionSocketConfig {
1204            frame_size: FRAME_SIZE,
1205            frame_timeout: Duration::from_millis(55),
1206            ..Default::default()
1207        };
1208
1209        let bob_cfg = SessionSocketConfig {
1210            frame_size: FRAME_SIZE,
1211            frame_timeout: Duration::from_millis(55),
1212            ..Default::default()
1213        };
1214
1215        #[cfg(feature = "telemetry")]
1216        let (alice_tracker, bob_tracker) = (TestTelemetryTracker::default(), TestTelemetryTracker::default());
1217
1218        let mut alice_socket = SessionSocket::<MTU, _>::new_stateless(
1219            "alice",
1220            alice,
1221            alice_cfg,
1222            #[cfg(feature = "telemetry")]
1223            alice_tracker.clone(),
1224        )?;
1225        let mut bob_socket = SessionSocket::<MTU, _>::new_stateless(
1226            "bob",
1227            bob,
1228            bob_cfg,
1229            #[cfg(feature = "telemetry")]
1230            bob_tracker.clone(),
1231        )?;
1232
1233        let alice_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
1234        alice_socket
1235            .write_all(&alice_sent_data)
1236            .timeout(futures_time::time::Duration::from_secs(2))
1237            .await??;
1238        alice_socket.flush().await?;
1239
1240        let bob_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
1241        bob_socket
1242            .write_all(&bob_sent_data)
1243            .timeout(futures_time::time::Duration::from_secs(2))
1244            .await??;
1245        bob_socket.flush().await?;
1246
1247        alice_socket.close().await?;
1248        bob_socket.close().await?;
1249
1250        let mut alice_recv_data = Vec::with_capacity(DATA_SIZE);
1251        alice_socket
1252            .read_to_end(&mut alice_recv_data)
1253            .timeout(futures_time::time::Duration::from_secs(2))
1254            .await??;
1255
1256        let mut bob_recv_data = Vec::with_capacity(DATA_SIZE);
1257        bob_socket
1258            .read_to_end(&mut bob_recv_data)
1259            .timeout(futures_time::time::Duration::from_secs(2))
1260            .await??;
1261
1262        // The whole first frame is discarded due to the missing first segment
1263        assert_eq!(bob_sent_data.len() - 1500, alice_recv_data.len());
1264        assert_eq!(&bob_sent_data[1500..], &alice_recv_data);
1265
1266        assert_eq!(alice_sent_data.len() - 1500, bob_recv_data.len());
1267        assert_eq!(&alice_sent_data[1500..], &bob_recv_data);
1268
1269        #[cfg(feature = "telemetry")]
1270        {
1271            insta::assert_yaml_snapshot!(alice_tracker);
1272            insta::assert_yaml_snapshot!(bob_tracker);
1273        }
1274
1275        Ok(())
1276    }
1277
1278    //#[test_log::test(tokio::test)]
1279    #[test_log::test(tokio::test(flavor = "multi_thread"))]
1280    async fn stateful_socket_bidirectional_should_should_not_skip_missing_frames() -> anyhow::Result<()> {
1281        let (alice, bob) = setup_alice_bob::<MTU>(
1282            FaultyNetworkConfig {
1283                avg_delay: Duration::from_millis(10),
1284                ids_to_drop: HashSet::from_iter([0_usize]),
1285                ..Default::default()
1286            },
1287            None,
1288            None,
1289        );
1290
1291        // use hopr_network_types::capture::PcapIoExt;
1292        // let (alice, bob) = (alice.capture("alice.pcap"), bob.capture("bob.pcap"));
1293
1294        let alice_cfg = SessionSocketConfig {
1295            frame_size: FRAME_SIZE,
1296            frame_timeout: Duration::from_millis(1000),
1297            ..Default::default()
1298        };
1299
1300        let bob_cfg = SessionSocketConfig {
1301            frame_size: FRAME_SIZE,
1302            frame_timeout: Duration::from_millis(1000),
1303            ..Default::default()
1304        };
1305
1306        let ack_cfg = AcknowledgementStateConfig {
1307            expected_packet_latency: Duration::from_millis(10),
1308            acknowledgement_delay: Duration::from_millis(40),
1309            ..Default::default()
1310        };
1311
1312        let (mut alice_rx, mut alice_tx) = SessionSocket::<MTU, _>::new(
1313            alice,
1314            AcknowledgementState::new("alice", ack_cfg),
1315            alice_cfg,
1316            #[cfg(feature = "telemetry")]
1317            NoopTracker,
1318        )?
1319        .split();
1320
1321        let (mut bob_rx, mut bob_tx) = SessionSocket::<MTU, _>::new(
1322            bob,
1323            AcknowledgementState::new("bob", ack_cfg),
1324            bob_cfg,
1325            #[cfg(feature = "telemetry")]
1326            NoopTracker,
1327        )?
1328        .split();
1329
1330        let alice_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
1331        let (alice_data_tx, alice_recv_data) = futures::channel::oneshot::channel();
1332        let alice_rx_jh = tokio::spawn(async move {
1333            let mut alice_recv_data = vec![0u8; DATA_SIZE];
1334            alice_rx.read_exact(&mut alice_recv_data).await?;
1335            alice_data_tx
1336                .send(alice_recv_data)
1337                .map_err(|_| std::io::Error::other("tx error"))?;
1338
1339            // Keep reading until the socket is closed
1340            alice_rx.read_to_end(&mut Vec::new()).await?;
1341            Ok::<_, std::io::Error>(())
1342        });
1343
1344        let bob_sent_data = hopr_types::crypto_random::random_bytes::<DATA_SIZE>();
1345        let (bob_data_tx, bob_recv_data) = futures::channel::oneshot::channel();
1346        let bob_rx_jh = tokio::spawn(async move {
1347            let mut bob_recv_data = vec![0u8; DATA_SIZE];
1348            bob_rx.read_exact(&mut bob_recv_data).await?;
1349            bob_data_tx
1350                .send(bob_recv_data)
1351                .map_err(|_| std::io::Error::other("tx error"))?;
1352
1353            // Keep reading until the socket is closed
1354            bob_rx.read_to_end(&mut Vec::new()).await?;
1355            Ok::<_, std::io::Error>(())
1356        });
1357
1358        let alice_tx_jh = tokio::spawn(async move {
1359            alice_tx
1360                .write_all(&alice_sent_data)
1361                .timeout(futures_time::time::Duration::from_secs(2))
1362                .await??;
1363            alice_tx.flush().await?;
1364
1365            // Once all data is sent, wait for the other side to receive it and close the socket
1366            let out = alice_recv_data.await.map_err(|_| std::io::Error::other("rx error"))?;
1367            alice_tx.close().await?;
1368            tracing::info!("alice closed");
1369            Ok::<_, std::io::Error>(out)
1370        });
1371
1372        let bob_tx_jh = tokio::spawn(async move {
1373            bob_tx
1374                .write_all(&bob_sent_data)
1375                .timeout(futures_time::time::Duration::from_secs(2))
1376                .await??;
1377            bob_tx.flush().await?;
1378
1379            // Once all data is sent, wait for the other side to receive it and close the socket
1380            let out = bob_recv_data.await.map_err(|_| std::io::Error::other("rx error"))?;
1381            bob_tx.close().await?;
1382            tracing::info!("bob closed");
1383            Ok::<_, std::io::Error>(out)
1384        });
1385
1386        let (alice_recv_data, bob_recv_data, a, b) =
1387            futures::future::try_join4(alice_tx_jh, bob_tx_jh, alice_rx_jh, bob_rx_jh)
1388                .timeout(futures_time::time::Duration::from_secs(4))
1389                .await??;
1390
1391        assert_eq!(&alice_sent_data, bob_recv_data?.as_slice());
1392        assert_eq!(&bob_sent_data, alice_recv_data?.as_slice());
1393        assert!(a.is_ok());
1394        assert!(b.is_ok());
1395
1396        Ok(())
1397    }
1398}
hopr_protocol_session/socket/mod.rs

hopr_protocol_session/socket/
mod.rs
