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

hopr_protocol_session/socket/
mod.rs
