hopr_protocol_session/socket/
ack_state.rs

1//! This module defines the [`SocketState`] that turns [`SessionSocket`](super::SessionSocket) into
2//! a reliable socket, with segment/frame retransmission and frame acknowledgements.
3
4use std::{
5    sync::atomic::AtomicBool,
6    time::{Duration, Instant},
7};
8
9use futures::{FutureExt, StreamExt};
10use futures_time::stream::StreamExt as TimeStreamExt;
11use hopr_utils::network_types::crossfire_sink::{CrossfireSink, bounded_sink_channel};
12use tracing::Instrument;
13
14use crate::{
15    errors::SessionError,
16    flow_control::DeliveryTap,
17    processing::types::FrameInspector,
18    protocol::{FrameAcknowledgements, FrameId, Segment, SegmentId, SegmentRequest, SeqIndicator, SessionMessage},
19    socket::{SocketState, state::SocketComponents},
20    utils::{
21        RetriedFrameId, RingBufferProducer, RingBufferView, next_deadline_with_backoff, searchable_ringbuffer,
22        skip_queue::{Skip, SkipDelaySender, skip_delay_channel},
23    },
24};
25
26/// Indicates the acknowledgement mode of a [stateful](AcknowledgementState) Session socket.
27#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
28pub enum AcknowledgementMode {
29    /// Partial frames are acknowledged, leading to receiver-driven retransmission requests.
30    /// The frame sender is never going to retransmit the entire frame.
31    Partial,
32    /// Only full frames are acknowledged, leading to the full-frame retransmission if
33    /// no acknowledgement is received by the frame sender.
34    Full,
35    /// Both partial and full acknowledgements are sent by the receiver.
36    ///
37    /// If a frame is partially acknowledged first, only receiver-driven retransmission requests follow
38    /// (as with [`AcknowledgementMode::Partial`].
39    ///
40    /// If the frame sender receives no acknowledgement (partial nor full), it retransmits
41    /// the entire frame (as in [`AcknowledgementMode::Full`]).
42    #[default]
43    Both,
44}
45
46impl AcknowledgementMode {
47    /// Indicates if `self` is [`AcknowledgementMode::Partial`] or [`AcknowledgementMode::Both`].
48    #[inline]
49    fn is_partial_ack_enabled(&self) -> bool {
50        matches!(self, Self::Partial | Self::Both)
51    }
52
53    /// Indicates if `self` is [`AcknowledgementMode::Full`] or [`AcknowledgementMode::Both`].
54    #[inline]
55    fn is_full_ack_enabled(&self) -> bool {
56        matches!(self, Self::Full | Self::Both)
57    }
58}
59
60/// Configuration object of the [`AcknowledgementState`].
61#[derive(Debug, Clone, Copy, PartialEq, smart_default::SmartDefault)]
62pub struct AcknowledgementStateConfig {
63    /// Mode of frame acknowledgement.
64    ///
65    /// Default is [`AcknowledgementMode::Both`]
66    pub mode: AcknowledgementMode,
67
68    /// The expected (average) latency of a packet (= single frame segment).
69    ///
70    /// Default is 20 ms
71    #[default(Duration::from_millis(20))]
72    pub expected_packet_latency: Duration,
73
74    /// Backoff base applied for segment or frame retransmissions.
75    ///
76    /// Default is 1.2
77    #[default(1.2)]
78    pub backoff_base: f64,
79
80    /// The maximum number of receiver-driven segment retransmission requests.
81    ///
82    /// Default is 3
83    #[default(3)]
84    pub max_incoming_frame_retries: usize,
85
86    /// The maximum number of sender-driven full-frame retransmissions.
87    ///
88    /// Default is 3
89    #[default(3)]
90    pub max_outgoing_frame_retries: usize,
91
92    /// Delay between acknowledgement batches.
93    ///
94    /// Default is 50 ms
95    #[default(Duration::from_millis(50))]
96    pub acknowledgement_delay: Duration,
97
98    /// Number of segments to hold back for retransmission upon other party's request.
99    /// Minimum is 1024.
100    ///
101    /// Default is 16 384.
102    #[default(16384)]
103    pub lookbehind_segments: usize,
104
105    /// Maximum age of an outgoing frame before it is retired as lost instead of retransmitted.
106    ///
107    /// Bounds how stale data on the wire may be after a stall, and feeds the retirement to the
108    /// honest delivery clock so flow control sees the stall as loss.
109    ///
110    /// `None` (default) bounds frames only by [`Self::max_outgoing_frame_retries`] and backoff.
111    #[default(None)]
112    pub max_frame_age: Option<Duration>,
113}
114
115impl AcknowledgementStateConfig {
116    fn normalize(self) -> AcknowledgementStateConfig {
117        Self {
118            mode: self.mode,
119            expected_packet_latency: self.expected_packet_latency.max(Duration::from_millis(1)),
120            backoff_base: self.backoff_base.max(1.0),
121            max_incoming_frame_retries: self.max_incoming_frame_retries,
122            max_outgoing_frame_retries: self.max_outgoing_frame_retries,
123            acknowledgement_delay: self.acknowledgement_delay.max(Duration::from_millis(1)),
124            lookbehind_segments: self.lookbehind_segments.max(1024),
125            // A zero bound would retire every frame on sight; treat it as "not set".
126            max_frame_age: self.max_frame_age.filter(|age| !age.is_zero()),
127        }
128    }
129}
130
131#[derive(Clone)]
132struct AcknowledgementStateContext<const C: usize> {
133    rb_tx: RingBufferProducer<Segment>,
134    rb_rx: RingBufferView<Segment>,
135    incoming_frame_retries_tx: SkipDelaySender<RetriedFrameId>,
136    outgoing_frame_retries_tx: SkipDelaySender<RetriedFrameId>,
137    ack_tx: CrossfireSink<FrameId>,
138    inspector: FrameInspector,
139    ctl_tx: CrossfireSink<SessionMessage<C>>,
140}
141
142#[cfg_attr(doc, aquamarine::aquamarine)]
143/// Represents a Session socket state is able to process acknowledgements.
144///
145/// # Retransmission driven by the Receiver
146/// ```mermaid
147/// sequenceDiagram
148///     Note over Sender,Receiver: Frame 1
149///     rect rgb(191, 223, 255)
150///     Note left of Sender: Frame 1 in buffer
151///     Sender->>Receiver: Segment 1/3 of Frame 1
152///     Sender->>Receiver: Segment 2/3 of Frame 1
153///     Sender--xReceiver: Segment 3/3 of Frame 1
154///     Note right of Receiver: RTO_BASE_RECEIVER elapsed
155///     Receiver->>Sender: Request Segment 3 of Frame 1
156///     Sender->>Receiver: Segment 3/3 of Frame 1
157///     Receiver->>Sender: Acknowledge Frame 1
158///     Note left of Sender: Frame 1 dropped from buffer
159///     end
160///     Note over Sender,Receiver: Frame 1 delivered
161/// ```
162///
163/// # Retransmission driven by the Sender
164/// ```mermaid
165/// sequenceDiagram
166///     Note over Sender,Receiver: Frame 1
167///     rect rgb(191, 223, 255)
168///     Note left of Sender: Frame 1 in buffer
169///     Sender->>Receiver: Segment 1/3 of Frame 1
170///     Sender->>Receiver: Segment 2/3 of Frame 1
171///     Sender--xReceiver: Segment 3/3 of Frame 1
172///     Note right of Receiver: RTO_BASE_RECEIVER elapsed
173///     Receiver--xSender: Request Segment 3 of Frame 1
174///     Note left of Sender: RTO_BASE_SENDER elapsed
175///     Sender->>Receiver: Segment 1/3 of Frame 1
176///     Sender->>Receiver: Segment 2/3 of Frame 1
177///     Sender->>Receiver: Segment 3/3 of Frame 1
178///     Receiver->>Sender: Acknowledge Frame 1
179///     Note left of Sender: Frame 1 dropped from buffer
180///     end
181///     Note over Sender,Receiver: Frame 1 delivered
182/// ```
183///
184/// # Sender-Receiver retransmission handover
185///
186/// ```mermaid
187///    sequenceDiagram
188///     Note over Sender,Receiver: Frame 1
189///     rect rgb(191, 223, 255)
190///     Note left of Sender: Frame 1 in buffer
191///     Sender->>Receiver: Segment 1/3 of Frame 1
192///     Sender--xReceiver: Segment 2/3 of Frame 1
193///     Sender--xReceiver: Segment 3/3 of Frame 1
194///     Note right of Receiver: RTO_BASE_RECEIVER elapsed
195///     Receiver->>Sender: Request Segments 2,3 of Frame 1
196///     Note left of Sender: RTO_BASE_SENDER cancelled
197///     Sender->>Receiver: Segment 2/3 of Frame 1
198///     Sender--xReceiver: Segment 3/3 of Frame 1
199///     Note right of Receiver: RTO_BASE_RECEIVER elapsed
200///     Receiver--xSender: Request Segments 3 of Frame 1
201///     Note right of Receiver: RTO_BASE_RECEIVER elapsed
202///     Receiver->>Sender: Request Segments 3 of Frame 1
203///     Sender->>Receiver: Segment 3/3 of Frame 1
204///     Receiver->>Sender: Acknowledge Frame 1
205///     Note left of Sender: Frame 1 dropped from buffer
206///     end
207///     Note over Sender,Receiver: Frame 1 delivered
208/// ```
209///
210/// # Retransmission failure
211///
212/// ```mermaid
213///    sequenceDiagram
214///     Note over Sender,Receiver: Frame 1
215///     rect rgb(191, 223, 255)
216///     Note left of Sender: Frame 1 in buffer
217///     Sender->>Receiver: Segment 1/3 of Frame 1
218///     Sender->>Receiver: Segment 2/3 of Frame 1
219///     Sender--xReceiver: Segment 3/3 of Frame 1
220///     Note right of Receiver: RTO_BASE_RECEIVER elapsed
221///     Receiver--xSender: Request Segment 3 of Frame 1
222///     Note left of Sender: RTO_BASE_SENDER elapsed
223///     Sender--xReceiver: Segment 1/3 of Frame 1
224///     Sender--xReceiver: Segment 2/3 of Frame 1
225///     Sender--xReceiver: Segment 3/3 of Frame 1
226///     Note left of Sender: FRAME_MAX_AGE elapsed<br/>Frame 1 dropped from buffer
227///     Note right of Receiver: FRAME_MAX_AGE elapsed<br/>Frame 1 dropped from buffer
228///     end
229///     Note over Sender,Receiver: Frame 1 never delivered
230/// ```
231#[derive(Clone)]
232pub struct AcknowledgementState<const C: usize> {
233    id: String,
234    cfg: AcknowledgementStateConfig,
235    context: Option<AcknowledgementStateContext<C>>,
236    started: std::sync::Arc<AtomicBool>,
237    /// Optional honest-clock tap for client-side flow control. Purely observational: acknowledged
238    /// and retransmission-exhausted frames bump the tap's atomic meter, but acknowledgement and
239    /// retransmission behaviour is unchanged whether or not it is installed.
240    delivery_tap: Option<DeliveryTap>,
241}
242
243impl<const C: usize> AcknowledgementState<C> {
244    pub fn new<I: std::fmt::Display>(session_id: I, cfg: AcknowledgementStateConfig) -> Self {
245        Self {
246            id: session_id.to_string(),
247            cfg: cfg.normalize(),
248            context: Default::default(),
249            started: std::sync::Arc::new(AtomicBool::new(false)),
250            delivery_tap: None,
251        }
252    }
253
254    /// Installs a read-only honest-clock tap. Each acknowledged frame bumps the tap's acked-bytes
255    /// counter; each frame whose sender-side retransmissions are exhausted bumps its lost-bytes
256    /// counter. This drives the client-side flow-control window ([`crate::flow_control`]) and never
257    /// alters delivery semantics.
258    pub fn with_delivery_tap(mut self, tap: DeliveryTap) -> Self {
259        self.delivery_tap = Some(tap);
260        self
261    }
262}
263
264impl<const C: usize> SocketState<C> for AcknowledgementState<C> {
265    fn session_id(&self) -> &str {
266        &self.id
267    }
268
269    #[tracing::instrument(name = "AcknowledgementState", skip(self, socket_components), fields(session_id = self.id))]
270    fn run(&mut self, socket_components: SocketComponents<C>) -> Result<(), SessionError> {
271        if self.started.load(std::sync::atomic::Ordering::Relaxed) && self.context.is_some() {
272            return Err(SessionError::InvalidState("state is already running".into()));
273        }
274
275        let (incoming_frame_retries_tx, incoming_frame_retries_rx) = skip_delay_channel();
276        let (outgoing_frame_retries_tx, outgoing_frame_retries_rx) = skip_delay_channel();
277        let (rb_tx, rb_rx) = searchable_ringbuffer(self.cfg.lookbehind_segments);
278
279        // Full frame acknowledgements get a special channel with fixed capacity
280        let (ack_tx, ack_rx) = bounded_sink_channel::<FrameId>(2 * self.cfg.lookbehind_segments);
281
282        let context = self.context.insert(AcknowledgementStateContext {
283            rb_tx,
284            rb_rx,
285            incoming_frame_retries_tx,
286            outgoing_frame_retries_tx,
287            ack_tx,
288            ctl_tx: socket_components.ctl_tx,
289            inspector: socket_components
290                .inspector
291                .ok_or(SessionError::InvalidState("inspector is not available".into()))?,
292        });
293
294        if self.cfg.mode.is_partial_ack_enabled() {
295            // For partially received frames incomplete for too long,
296            // missing segments will be asked for retransmission
297            let mut incoming_frame_retries_tx_clone = context.incoming_frame_retries_tx.clone();
298            let ctl_tx_clone = context.ctl_tx.clone();
299            let frame_inspector_clone = context.inspector.clone();
300            let cfg = self.cfg;
301            hopr_utils::runtime::prelude::spawn(incoming_frame_retries_rx
302                .filter_map(move |rf| {
303                    let frame_id = rf.frame_id;
304                    let missing_segments = frame_inspector_clone.missing_segments(&frame_id).unwrap_or_default();
305                    if !missing_segments.is_empty() {
306                        // Find out if we need to subscribe for further retries of this Frame
307                        if let Some(next) = rf.next() {
308                            // Register the next retry if still possible
309                            let retry_at = next_deadline_with_backoff(next.retry_count, cfg.backoff_base, cfg.expected_packet_latency);
310                            if let Err(error) = incoming_frame_retries_tx_clone.send_one((next, retry_at)) {
311                                tracing::error!(frame_id, %error, "failed to register next resend of incoming frame");
312                            } else {
313                                tracing::debug!(frame_id, retry_in = ?retry_at.saturating_duration_since(Instant::now()), "next resend request of incoming frame segments");
314                            }
315                        } else {
316                            tracing::debug!(frame_id, "last request of incoming frame segments");
317                        }
318
319                        futures::future::ready(Some((frame_id, missing_segments)))
320                    } else {
321                        tracing::debug!(frame_id, "no more missing segments in frame");
322                        futures::future::ready(None)
323                    }
324                })
325                .ready_chunks(SegmentRequest::<C>::MAX_ENTRIES)
326                .inspect(|r| tracing::trace!(req = ?r, "requesting segments resend"))
327                .map(|a| Ok(SessionMessage::<C>::Request(a.into_iter().collect())))
328                .forward(ctl_tx_clone)
329                .map(move |res| match res {
330                    Ok(_) => tracing::debug!("incoming frame resends processing done"),
331                    Err(error) => tracing::error!(%error, "error while processing incoming frame resends")
332                })
333                .instrument(tracing::debug_span!("incoming_frame_retries_sender"))
334            );
335        }
336
337        // Send out Frame Acknowledgements chunked as Control messages
338        let ctl_tx_clone = context.ctl_tx.clone();
339        let ack_delay = self.cfg.acknowledgement_delay;
340        hopr_utils::runtime::prelude::spawn(
341            ack_rx
342                .buffer(futures_time::time::Duration::from(ack_delay))
343                .flat_map(|acks| futures::stream::iter(FrameAcknowledgements::<C>::new_multiple(acks)))
344                .filter(|acks| futures::future::ready(!acks.is_empty()))
345                .inspect(|acks| tracing::trace!(?acks, "acknowledgements sent"))
346                .map(|acks| Ok(SessionMessage::<C>::Acknowledge(acks)))
347                .forward(ctl_tx_clone)
348                .map(move |res| match res {
349                    Ok(_) => tracing::debug!("acknowledgement forwarding done"),
350                    Err(error) => tracing::debug!(%error, "acknowledgement forwarding failed"),
351                })
352                .instrument(tracing::debug_span!("acknowledgement_sender")),
353        );
354
355        // Resend outgoing frame Segments if they were not (partially or fully) acknowledged
356        let mut outgoing_frame_retries_tx_clone = context.outgoing_frame_retries_tx.clone();
357        let ctl_tx_clone = context.ctl_tx.clone();
358        let rb_rx_clone = context.rb_rx.clone();
359        let cfg = self.cfg;
360        let delivery_tap = self.delivery_tap.clone();
361        hopr_utils::runtime::prelude::spawn(
362            outgoing_frame_retries_rx
363                .filter_map(move |rf: RetriedFrameId| {
364                    let frame_id = rf.frame_id;
365
366                    // Anti-bufferbloat: data this stale is worthless to the peer, but the latency
367                    // it would add on a burst drain is not. Retire it as lost instead — which also
368                    // tells the honest clock the truth about the stall.
369                    if cfg.max_frame_age.is_some_and(|max_age| rf.age() >= max_age) {
370                        tracing::trace!(frame_id, age = ?rf.age(), "outgoing frame exceeded max age; retiring as lost");
371                        if let Some(tap) = &delivery_tap {
372                            tap.on_lost_frame();
373                        }
374                        return futures::future::ready(None);
375                    }
376
377                    // Find out if the frame can be retried again in the future
378                    if let Some(next) = rf.next() {
379                        // Register the next retry if still possible
380                        let retry_at =
381                            next_deadline_with_backoff(next.retry_count, cfg.backoff_base, cfg.expected_packet_latency);
382                        if let Err(error) = outgoing_frame_retries_tx_clone.send_one((next, retry_at)) {
383                            tracing::error!(frame_id, %error, "failed to register next retry of frame");
384                        } else {
385                            tracing::debug!(frame_id, retry_in = ?retry_at.saturating_duration_since(Instant::now()), "next resend of outgoing frame");
386                        }
387                    } else {
388                        tracing::debug!(frame_id, "last outgoing retry of frame");
389                        // Honest-clock tap: retransmissions exhausted → treat the frame as lost.
390                        // An earlier ack cancels this retry (Skip), so a delivered frame never
391                        // reaches here; a late ack after give-up is harmless (bounded over-credit).
392                        if let Some(tap) = &delivery_tap {
393                            tap.on_lost_frame();
394                        }
395                    }
396                    tracing::trace!(frame_id, "going to re-send entire frame");
397                    futures::future::ready(Some(frame_id))
398                })
399                .flat_map(move |frame_id| {
400                    // Find out all the segments of that frame to be retransmitted
401                    futures::stream::iter(
402                        rb_rx_clone
403                            .find(|s: &Segment| s.id().0 == frame_id)
404                            .into_iter()
405                            .inspect(|s| tracing::trace!(seg_id = %s.id(), "segment retransmit"))
406                            .map(|s| Ok(SessionMessage::<C>::Segment(s))),
407                    )
408                })
409                .forward(ctl_tx_clone) // Retransmit all the segments
410                .map(move |res| match res {
411                    Ok(_) => tracing::debug!("outgoing frame retries processing done"),
412                    Err(error) => tracing::error!(%error, "error while processing outgoing frame retries"),
413                })
414                .instrument(tracing::debug_span!("outgoing_frame_retries_sender")),
415        );
416
417        tracing::debug!("acknowledgement state has been started");
418        self.started.store(true, std::sync::atomic::Ordering::Relaxed);
419
420        Ok(())
421    }
422
423    #[tracing::instrument(name = "AcknowledgementState::stop", skip(self), fields(session_id = self.id))]
424    fn stop(&mut self) -> Result<(), SessionError> {
425        if let Some(mut ctx) = self.context.take() {
426            ctx.outgoing_frame_retries_tx.force_close();
427            ctx.incoming_frame_retries_tx.force_close();
428            // ack_tx and ctl_tx close when their last clone is dropped;
429            // dropping ctx here drops the originals, spawned-task clones drop when those tasks complete.
430
431            self.started.store(false, std::sync::atomic::Ordering::Relaxed);
432            tracing::debug!("state has been stopped");
433        } else {
434            tracing::warn!("cannot be stopped, because it is not running");
435        }
436
437        Ok(())
438    }
439
440    #[tracing::instrument(name = "AcknowledgementState::incoming_segment", skip(self), fields(session_id = self.id, frame_id = seg_id.0))]
441    fn incoming_segment(&mut self, seg_id: &SegmentId, _ind: SeqIndicator) -> Result<(), SessionError> {
442        tracing::trace!("segment received");
443
444        let ctx = self
445            .started
446            .load(std::sync::atomic::Ordering::Relaxed)
447            .then_some(self.context.as_mut())
448            .flatten()
449            .ok_or(SessionError::StateNotRunning)?;
450
451        // Register future requesting of segments for this frame
452        if self.cfg.mode.is_partial_ack_enabled() {
453            // Every incoming segment of this frame will move the deadline further
454            // into the future.
455            if let Err(error) = ctx.incoming_frame_retries_tx.send_one((
456                RetriedFrameId::with_retries(seg_id.0, self.cfg.max_incoming_frame_retries),
457                self.cfg.expected_packet_latency, // RTO_BASE_RECEIVER - when we expect the next segment to arrive
458            )) {
459                tracing::error!(%error, "failed to register incoming retry for frame");
460            }
461        }
462        Ok(())
463    }
464
465    #[tracing::instrument(name = "AcknowledgementState::incoming_retransmission_request", skip(self, request), fields(session_id = self.id))]
466    fn incoming_retransmission_request(&mut self, request: SegmentRequest<C>) -> Result<(), SessionError> {
467        // The state will respond to segment retransmission requests even
468        // if it has this feature disabled in the config.
469        tracing::trace!(count = request.len(), "segment retransmission requested");
470
471        let ctx = self
472            .started
473            .load(std::sync::atomic::Ordering::Relaxed)
474            .then_some(self.context.as_mut())
475            .flatten()
476            .ok_or(SessionError::StateNotRunning)?;
477
478        let (mut missing_seg_ids, mut missing_frame_ids): (Vec<_>, Vec<_>) =
479            request.into_iter().map(|s| (s, s.0)).unzip();
480
481        // Perform a single find to lock the RB only once
482        let segments = ctx.rb_rx.find(|s| {
483            // SegmentIds are guaranteed to be sorted, so we can use binary search
484            if let Ok(i) = missing_seg_ids.binary_search(&s.id()) {
485                missing_seg_ids.remove(i);
486                true
487            } else {
488                false
489            }
490        });
491
492        tracing::trace!(
493            found = segments.len(),
494            requested = missing_frame_ids.len(),
495            "found matching segments to be retransmitted"
496        );
497
498        // Partially acknowledged frames will not need to be fully resent in the future.
499        // Cancel all partially acknowledged frame resends.
500        if self.cfg.mode.is_full_ack_enabled() {
501            // Since the FrameIds are guaranteed to be sorted, we can simply dedup them.
502            missing_frame_ids.dedup();
503
504            if let Err(error) = ctx.outgoing_frame_retries_tx.send_many(
505                missing_frame_ids
506                    .into_iter()
507                    .map(|frame_id| (RetriedFrameId::no_retries(frame_id), Skip).into()),
508            ) {
509                tracing::error!(%error, "failed to cancel frame resend of partially acknowledged frames");
510            }
511        }
512
513        // Resend the segments via the Control Stream
514        segments
515            .into_iter()
516            .try_for_each(|s| {
517                tracing::trace!(seg_id = %s.id(), "retransmit segment on request");
518                ctx.ctl_tx.try_send(SessionMessage::Segment(s))
519            })
520            .map_err(|e| SessionError::ProcessingError(e.to_string()))
521    }
522
523    #[tracing::instrument(name = "AcknowledgementState::incoming_acknowledged_frames", skip(self), fields(session_id = self.id))]
524    fn incoming_acknowledged_frames(&mut self, ack: FrameAcknowledgements<C>) -> Result<(), SessionError> {
525        tracing::trace!(count = ack.len(), "frame acknowledgements received");
526
527        // Cloned out before the mutable `ctx` borrow so the honest-clock tap can run inside the
528        // acknowledgement iteration without conflicting with the retry-cancel borrow.
529        let tap = self.delivery_tap.clone();
530        let full_ack = self.cfg.mode.is_full_ack_enabled();
531
532        let ctx = self
533            .started
534            .load(std::sync::atomic::Ordering::Relaxed)
535            .then_some(self.context.as_mut())
536            .flatten()
537            .ok_or(SessionError::StateNotRunning)?;
538
539        // Honest-clock tap: every received frame acknowledgement is proof of delivery, and must feed
540        // the flow-control window regardless of the ack mode. `full_ack` only governs whether we
541        // *also* cancel the outgoing full-frame retransmission below (in `Partial` mode the receiver
542        // drives retransmission, so there is nothing to cancel here — but the acks still arrive).
543        if let Some(tap) = &tap {
544            (0..ack.len()).for_each(|_| tap.on_acked_frame());
545        }
546
547        // Frame acknowledged, we will not need to resend it (full-ack mode only).
548        if full_ack
549            && let Err(error) = ctx.outgoing_frame_retries_tx.send_many(
550                ack.into_iter()
551                    .inspect(|frame_id| tracing::trace!(frame_id, "frame acknowledged"))
552                    .map(|frame_id| (RetriedFrameId::no_retries(frame_id), Skip).into()),
553            )
554        {
555            tracing::error!(%error, "failed to cancel frame resend");
556        }
557
558        Ok(())
559    }
560
561    #[tracing::instrument(name = "AcknowledgementState::frame_complete", skip(self), fields(session_id = self.id))]
562    fn frame_complete(&mut self, frame_id: FrameId) -> Result<(), SessionError> {
563        tracing::trace!("frame complete");
564
565        let ctx = self
566            .started
567            .load(std::sync::atomic::Ordering::Relaxed)
568            .then_some(self.context.as_mut())
569            .flatten()
570            .ok_or(SessionError::StateNotRunning)?;
571
572        // Since the frame has been completed, push its ID into the acknowledgement queue
573        if let Err(error) = ctx.ack_tx.try_send(frame_id) {
574            tracing::error!(%error, "failed to acknowledge frame");
575        }
576
577        if self.cfg.mode.is_partial_ack_enabled() {
578            // No more requesting of segment retransmissions from frames that were completed
579            if let Err(error) = ctx
580                .incoming_frame_retries_tx
581                .send_one((RetriedFrameId::no_retries(frame_id), Skip))
582            {
583                tracing::error!(%error, "failed to cancel retry of acknowledged frame");
584            }
585        }
586
587        Ok(())
588    }
589
590    #[tracing::instrument(name = "AcknowledgementState::frame_emitted", skip(self), fields(session_id = self.id))]
591    fn frame_emitted(&mut self, id: FrameId) -> Result<(), SessionError> {
592        tracing::trace!("frame emitted");
593        let _ = self
594            .started
595            .load(std::sync::atomic::Ordering::Relaxed)
596            .then_some(self.context.as_mut())
597            .flatten()
598            .ok_or(SessionError::StateNotRunning)?;
599        Ok(())
600    }
601
602    #[tracing::instrument(name = "AcknowledgementState::frame_discarded", skip(self), fields(session_id = self.id))]
603    fn frame_discarded(&mut self, frame_id: FrameId) -> Result<(), SessionError> {
604        tracing::trace!("frame discarded");
605
606        let ctx = self
607            .started
608            .load(std::sync::atomic::Ordering::Relaxed)
609            .then_some(self.context.as_mut())
610            .flatten()
611            .ok_or(SessionError::StateNotRunning)?;
612
613        if self.cfg.mode.is_partial_ack_enabled() {
614            // No more requesting of segment retransmissions from frames that were discarded
615            if let Err(error) = ctx
616                .incoming_frame_retries_tx
617                .send_one((RetriedFrameId::no_retries(frame_id), Skip))
618            {
619                tracing::error!(%error, "failed to cancel retry of acknowledged frame");
620            }
621        }
622
623        Ok(())
624    }
625
626    #[tracing::instrument(name = "AcknowledgementState::segment_sent", skip(self, segment), fields(session_id = self.id, frame_id = segment.frame_id, seq_idx = segment.seq_idx))]
627    fn segment_sent(&mut self, segment: &Segment) -> Result<(), SessionError> {
628        tracing::trace!("segment sent");
629
630        let ctx = self
631            .started
632            .load(std::sync::atomic::Ordering::Relaxed)
633            .then_some(self.context.as_mut())
634            .flatten()
635            .ok_or(SessionError::StateNotRunning)?;
636
637        // Since segments are re-sent via Control stream, they are not later fed again
638        // into the ring buffer.
639        if !ctx.rb_tx.push(segment.clone()) {
640            tracing::trace!(
641                frame_id = segment.frame_id,
642                seq_idx = segment.seq_idx,
643                "segment dropped: ring buffer full"
644            );
645        }
646
647        // When the last segment of a frame has been sent,
648        // add it to outgoing retries (if the full ack mode is enabled).
649        if segment.is_last() && self.cfg.mode.is_full_ack_enabled() {
650            tracing::trace!("last segment of frame sent");
651
652            if let Err(error) = ctx.outgoing_frame_retries_tx.send_one((
653                RetriedFrameId::with_retries(segment.frame_id, self.cfg.max_outgoing_frame_retries),
654                // The whole frame should be delivered and acknowledged
655                // once all its segments (seq_len) are sent,
656                // and the acknowledgement also comes back to us.
657                // Therefore, RTO_BASE_SENDER = latency * (seq_len + 1)
658                self.cfg.expected_packet_latency * (segment.seq_flags.seq_len() + 1) as u32,
659            )) {
660                tracing::error!(%error, "failed to insert outgoing retry of a frame");
661            }
662        }
663
664        Ok(())
665    }
666}
667
668#[cfg(test)]
669mod tests {
670    use anyhow::Context;
671
672    use super::*;
673    use crate::{
674        processing::types::{FrameBuilder, FrameDashMap, FrameMap},
675        protocol::SeqNum,
676        utils::segment,
677    };
678
679    const FRAME_SIZE: usize = 1500;
680
681    const MTU: usize = 1000;
682
683    #[test_log::test(tokio::test)]
684    async fn ack_state_sender_must_acknowledge_completed_frames() -> anyhow::Result<()> {
685        let cfg = AcknowledgementStateConfig {
686            expected_packet_latency: Duration::from_millis(10),
687            acknowledgement_delay: Duration::from_millis(2),
688            ..Default::default()
689        };
690
691        let inspector = FrameInspector(FrameDashMap::with_capacity(10));
692        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
693
694        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
695        state.run(SocketComponents {
696            inspector: inspector.into(),
697            ctl_tx,
698        })?;
699
700        let acked_frame_ids = [1, 2, 3];
701
702        for &frame_id in &acked_frame_ids {
703            state.frame_complete(frame_id)?;
704        }
705
706        tokio::time::sleep(cfg.acknowledgement_delay * 2).await;
707
708        state.stop()?;
709
710        let ctl_msgs = tokio::time::timeout(Duration::from_millis(100), ctl_rx.collect::<Vec<_>>())
711            .await
712            .context("timeout receiving Control messages")?;
713
714        assert_eq!(1, ctl_msgs.len());
715
716        assert_eq!(
717            ctl_msgs[0],
718            SessionMessage::Acknowledge(acked_frame_ids.to_vec().try_into()?)
719        );
720
721        Ok(())
722    }
723
724    #[parameterized::parameterized(num_frames = { 1, 2, 3 })]
725    #[parameterized_macro(test_log::test(tokio::test))]
726    async fn ack_state_sender_must_resend_unacknowledged_frames(num_frames: usize) -> anyhow::Result<()> {
727        const NUM_RETRIES: usize = 2;
728
729        let cfg = AcknowledgementStateConfig {
730            mode: AcknowledgementMode::Full,
731            expected_packet_latency: Duration::from_millis(2),
732            max_outgoing_frame_retries: NUM_RETRIES,
733            ..Default::default()
734        };
735
736        let inspector = FrameInspector(FrameDashMap::with_capacity(10));
737        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
738
739        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
740        state.run(SocketComponents {
741            inspector: inspector.into(),
742            ctl_tx,
743        })?;
744
745        let mut expected_frame_segments = Vec::new();
746        let num_segments_in_frame = FRAME_SIZE / MTU + 1;
747        for i in 1..=num_frames {
748            let expected_segments = segment(
749                hopr_types::crypto_random::random_bytes::<FRAME_SIZE>(),
750                MTU,
751                i as FrameId,
752            )?;
753            for segment in &expected_segments {
754                state.segment_sent(segment)?;
755            }
756            expected_frame_segments.push(expected_segments);
757        }
758
759        let expected_frame_delivery = cfg.expected_packet_latency * (num_segments_in_frame + 1) as u32;
760        tokio::time::sleep(2 * expected_frame_delivery).await;
761        state.stop()?;
762
763        let ctl_msg = tokio::time::timeout(Duration::from_millis(100), ctl_rx.collect::<Vec<_>>())
764            .await
765            .context("timeout receiving Control message")?;
766
767        let retransmitted_segments = ctl_msg
768            .into_iter()
769            .map(|m| m.try_as_segment().ok_or(anyhow::anyhow!("must be segment")))
770            .collect::<Result<Vec<_>, _>>()?;
771
772        assert_eq!(
773            NUM_RETRIES * num_segments_in_frame * num_frames,
774            retransmitted_segments.len()
775        );
776
777        let total_segments = expected_frame_segments.iter().map(|m| m.len()).sum::<usize>();
778        let expected_segments = expected_frame_segments
779            .into_iter()
780            .flatten()
781            .cycle()
782            .take(total_segments * NUM_RETRIES)
783            .collect::<Vec<_>>();
784        assert_eq!(expected_segments, retransmitted_segments);
785
786        Ok(())
787    }
788
789    /// Runs one unacknowledged frame through the retry pipeline and returns how many segments were
790    /// retransmitted before `run_for` elapsed.
791    async fn retransmissions_with_max_frame_age(max_frame_age: Option<Duration>) -> anyhow::Result<usize> {
792        // A generous retry budget, so the only thing that can cut the retries short is the age bound.
793        let cfg = AcknowledgementStateConfig {
794            mode: AcknowledgementMode::Full,
795            expected_packet_latency: Duration::from_millis(5),
796            max_outgoing_frame_retries: 20,
797            max_frame_age,
798            ..Default::default()
799        };
800
801        let inspector = FrameInspector(FrameDashMap::with_capacity(10));
802        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
803
804        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
805        state.run(SocketComponents {
806            inspector: inspector.into(),
807            ctl_tx,
808        })?;
809
810        for segment in &segment(hopr_types::crypto_random::random_bytes::<FRAME_SIZE>(), MTU, 1)? {
811            state.segment_sent(segment)?;
812        }
813
814        // Never acknowledge: the frame stays in the retry pipeline for the whole window.
815        tokio::time::sleep(Duration::from_millis(500)).await;
816        state.stop()?;
817
818        let ctl_msgs = tokio::time::timeout(Duration::from_millis(100), ctl_rx.collect::<Vec<_>>())
819            .await
820            .context("timeout receiving Control messages")?;
821
822        Ok(ctl_msgs.into_iter().filter_map(|m| m.try_as_segment()).count())
823    }
824
825    #[test_log::test(tokio::test)]
826    async fn ack_state_sender_should_stop_resending_a_frame_that_exceeded_the_max_age() -> anyhow::Result<()> {
827        let unbounded = retransmissions_with_max_frame_age(None).await?;
828        let bounded = retransmissions_with_max_frame_age(Some(Duration::from_millis(50))).await?;
829
830        assert!(
831            unbounded > 0,
832            "the retry pipeline must retransmit at all without an age bound"
833        );
834        assert!(
835            bounded < unbounded,
836            "an age bound must cut retransmissions short: got {bounded} bounded vs {unbounded} unbounded"
837        );
838
839        Ok(())
840    }
841
842    #[test_log::test(tokio::test)]
843    async fn ack_state_sender_must_not_resend_unacknowledged_frame_when_full_resend_disabled() -> anyhow::Result<()> {
844        const NUM_RETRIES: usize = 2;
845
846        let cfg = AcknowledgementStateConfig {
847            mode: AcknowledgementMode::Partial,
848            expected_packet_latency: Duration::from_millis(2),
849            max_outgoing_frame_retries: NUM_RETRIES,
850            ..Default::default()
851        };
852
853        let inspector = FrameInspector(FrameDashMap::with_capacity(10));
854        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
855
856        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
857        state.run(SocketComponents {
858            inspector: inspector.into(),
859            ctl_tx,
860        })?;
861
862        let expected_segments = segment(hopr_types::crypto_random::random_bytes::<FRAME_SIZE>(), MTU, 1)?;
863        for segment in &expected_segments {
864            state.segment_sent(segment)?;
865        }
866
867        let expected_frame_delivery = cfg.expected_packet_latency * (expected_segments.len() + 1) as u32;
868        tokio::time::sleep(2 * expected_frame_delivery).await;
869        state.stop()?;
870
871        // No retransmission should be sent because it is disabled
872        assert!(ctl_rx.collect::<Vec<_>>().await.is_empty());
873
874        Ok(())
875    }
876
877    #[tokio::test]
878    async fn ack_state_sender_must_not_resend_acknowledged_frame() -> anyhow::Result<()> {
879        let cfg = AcknowledgementStateConfig {
880            mode: AcknowledgementMode::Full,
881            expected_packet_latency: Duration::from_millis(2),
882            max_outgoing_frame_retries: 1,
883            ..Default::default()
884        };
885
886        let inspector = FrameInspector(FrameDashMap::with_capacity(10));
887        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
888
889        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
890        state.run(SocketComponents {
891            inspector: inspector.into(),
892            ctl_tx,
893        })?;
894
895        let expected_segments = segment(hopr_types::crypto_random::random_bytes::<{ FRAME_SIZE * 2 }>(), MTU, 1)?;
896        for segment in &expected_segments {
897            state.segment_sent(segment)?;
898        }
899
900        // Acknowledge the frame
901        state.incoming_acknowledged_frames(vec![1].try_into()?)?;
902
903        tokio::time::sleep(10 * cfg.expected_packet_latency).await;
904
905        state.stop()?;
906
907        // No retransmission should be sent because the frame was already acknowledged.
908        assert!(ctl_rx.collect::<Vec<_>>().await.is_empty());
909
910        Ok(())
911    }
912
913    #[test_log::test(tokio::test)]
914    async fn ack_state_sender_must_not_resend_entire_frame_when_already_partially_acknowledged() -> anyhow::Result<()> {
915        let cfg = AcknowledgementStateConfig {
916            mode: AcknowledgementMode::Full,
917            expected_packet_latency: Duration::from_millis(2),
918            max_outgoing_frame_retries: 1,
919            ..Default::default()
920        };
921
922        let inspector = FrameInspector(FrameDashMap::with_capacity(10));
923        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
924
925        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
926        state.run(SocketComponents {
927            inspector: inspector.into(),
928            ctl_tx,
929        })?;
930
931        let expected_segments = segment(hopr_types::crypto_random::random_bytes::<{ FRAME_SIZE * 2 }>(), MTU, 1)?;
932
933        // Load segments into the ring buffer
934        for segment in &expected_segments {
935            state.segment_sent(segment)?;
936        }
937
938        tokio::time::sleep(cfg.expected_packet_latency).await;
939
940        // Partially acknowledge the frame (report the first segment as missing)
941        state.incoming_retransmission_request(SegmentRequest::from_iter([(1, [0b10000000].into())]))?;
942
943        state.stop()?;
944
945        // Only segment 1 must be retransmitted
946        let ctl_msgs = ctl_rx.collect::<Vec<_>>().await;
947        assert_eq!(1, ctl_msgs.len());
948        assert_eq!(ctl_msgs[0], SessionMessage::Segment(expected_segments[0].clone()));
949
950        Ok(())
951    }
952
953    #[test_log::test(tokio::test)]
954    async fn ack_state_sender_must_retransmit_segments_when_requested() -> anyhow::Result<()> {
955        let cfg = AcknowledgementStateConfig {
956            mode: AcknowledgementMode::Full,
957            expected_packet_latency: Duration::from_millis(2),
958            max_outgoing_frame_retries: 1,
959            ..Default::default()
960        };
961
962        let inspector = FrameInspector(FrameDashMap::with_capacity(10));
963        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
964
965        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
966        state.run(SocketComponents {
967            inspector: inspector.into(),
968            ctl_tx,
969        })?;
970
971        let expected_segments_1 = segment(hopr_types::crypto_random::random_bytes::<{ FRAME_SIZE * 2 }>(), MTU, 1)?;
972        // Load frame 1 segments into the ring buffer
973        for segment in &expected_segments_1 {
974            state.segment_sent(segment)?;
975        }
976
977        let expected_segments_2 = segment(hopr_types::crypto_random::random_bytes::<{ FRAME_SIZE * 2 }>(), MTU, 2)?;
978        // Load frame 2 segments into the ring buffer
979        for segment in &expected_segments_2 {
980            state.segment_sent(segment)?;
981        }
982
983        tokio::time::sleep(cfg.expected_packet_latency).await;
984
985        // Request different segments to be retransmitted
986        state.incoming_retransmission_request(SegmentRequest::from_iter([
987            (1, [0b11100000].into()),
988            (2, [0b11100000].into()),
989        ]))?;
990        tokio::time::sleep(cfg.expected_packet_latency).await;
991
992        state.incoming_retransmission_request(SegmentRequest::from_iter([(2, [0b11000000].into())]))?;
993        tokio::time::sleep(cfg.expected_packet_latency).await;
994
995        state.incoming_retransmission_request(SegmentRequest::from_iter([(2, [0b01000000].into())]))?;
996        tokio::time::sleep(cfg.expected_packet_latency).await;
997
998        state.stop()?;
999
1000        let ctl_msgs = ctl_rx.collect::<Vec<_>>().await;
1001
1002        assert_eq!(9, ctl_msgs.len());
1003        // Request 1 - frame 1
1004        assert_eq!(ctl_msgs[0], SessionMessage::Segment(expected_segments_1[0].clone()));
1005        assert_eq!(ctl_msgs[1], SessionMessage::Segment(expected_segments_1[1].clone()));
1006        assert_eq!(ctl_msgs[2], SessionMessage::Segment(expected_segments_1[2].clone()));
1007        // Request 1 - frame 2
1008        assert_eq!(ctl_msgs[3], SessionMessage::Segment(expected_segments_2[0].clone()));
1009        assert_eq!(ctl_msgs[4], SessionMessage::Segment(expected_segments_2[1].clone()));
1010        assert_eq!(ctl_msgs[5], SessionMessage::Segment(expected_segments_2[2].clone()));
1011
1012        // Request 2 - frame 2
1013        assert_eq!(ctl_msgs[6], SessionMessage::Segment(expected_segments_2[0].clone()));
1014        assert_eq!(ctl_msgs[7], SessionMessage::Segment(expected_segments_2[1].clone()));
1015
1016        // Request 3 - frame 2
1017        assert_eq!(ctl_msgs[8], SessionMessage::Segment(expected_segments_2[1].clone()));
1018
1019        Ok(())
1020    }
1021
1022    #[tokio::test]
1023    async fn ack_state_receiver_must_request_missing_frames_when_partial_acks_are_enabled() -> anyhow::Result<()> {
1024        let cfg = AcknowledgementStateConfig {
1025            mode: AcknowledgementMode::Partial,
1026            expected_packet_latency: Duration::from_millis(2),
1027            max_incoming_frame_retries: 1,
1028            ..Default::default()
1029        };
1030
1031        let mut inspector = FrameInspector(FrameDashMap::with_capacity(10));
1032        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
1033
1034        let segments = segment(hopr_types::crypto_random::random_bytes::<FRAME_SIZE>(), MTU, 1)?;
1035
1036        inspector
1037            .0
1038            .entry(1)
1039            .try_as_vacant()
1040            .ok_or(anyhow::anyhow!("frame 1 must be vacant"))?
1041            .insert(FrameBuilder::from(segments[0].clone()));
1042
1043        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
1044        state.run(SocketComponents {
1045            inspector: inspector.into(),
1046            ctl_tx,
1047        })?;
1048
1049        state.incoming_segment(&segments[0].id(), (segments.len() as SeqNum).try_into()?)?;
1050
1051        tokio::time::sleep(cfg.expected_packet_latency * 2).await;
1052
1053        state.stop()?;
1054
1055        let ctl_msgs = tokio::time::timeout(Duration::from_millis(100), ctl_rx.collect::<Vec<_>>())
1056            .await
1057            .context("timeout receiving Control messages")?;
1058
1059        assert_eq!(1, ctl_msgs.len());
1060        assert_eq!(
1061            ctl_msgs[0],
1062            SessionMessage::Request(SegmentRequest::from_iter([(1, [0b01000000].into())]))
1063        );
1064
1065        Ok(())
1066    }
1067
1068    #[tokio::test]
1069    async fn ack_state_receiver_must_not_request_missing_frames_when_partial_acks_are_disabled() -> anyhow::Result<()> {
1070        let cfg = AcknowledgementStateConfig {
1071            mode: AcknowledgementMode::Full,
1072            expected_packet_latency: Duration::from_millis(2),
1073            max_incoming_frame_retries: 1,
1074            ..Default::default()
1075        };
1076
1077        let mut inspector = FrameInspector(FrameDashMap::with_capacity(10));
1078        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
1079
1080        let segments = segment(hopr_types::crypto_random::random_bytes::<FRAME_SIZE>(), MTU, 1)?;
1081
1082        inspector
1083            .0
1084            .entry(1)
1085            .try_as_vacant()
1086            .ok_or(anyhow::anyhow!("frame 1 must be vacant"))?
1087            .insert(FrameBuilder::from(segments[0].clone()));
1088
1089        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
1090        state.run(SocketComponents {
1091            inspector: inspector.into(),
1092            ctl_tx,
1093        })?;
1094
1095        state.incoming_segment(&segments[0].id(), (segments.len() as SeqNum).try_into()?)?;
1096
1097        tokio::time::sleep(cfg.expected_packet_latency * 2).await;
1098
1099        state.stop()?;
1100
1101        let ctl_msgs = tokio::time::timeout(Duration::from_millis(100), ctl_rx.collect::<Vec<_>>())
1102            .await
1103            .context("timeout receiving Control messages")?;
1104
1105        assert!(ctl_msgs.iter().all(|m| !matches!(m, SessionMessage::Request(_))));
1106
1107        Ok(())
1108    }
1109
1110    #[tokio::test]
1111    async fn ack_state_receiver_must_continue_requesting_missing_frames_when_frame_not_completed() -> anyhow::Result<()>
1112    {
1113        let cfg = AcknowledgementStateConfig {
1114            mode: AcknowledgementMode::Partial,
1115            expected_packet_latency: Duration::from_millis(2),
1116            max_incoming_frame_retries: 3,
1117            ..Default::default()
1118        };
1119
1120        let mut inspector = FrameInspector(FrameDashMap::with_capacity(10));
1121        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
1122
1123        let segments = segment(hopr_types::crypto_random::random_bytes::<{ 2 * FRAME_SIZE }>(), MTU, 1)?;
1124
1125        inspector
1126            .0
1127            .entry(1)
1128            .try_as_vacant()
1129            .ok_or(anyhow::anyhow!("frame 1 must be vacant"))?
1130            .insert(FrameBuilder::from(segments[0].clone()));
1131
1132        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
1133        state.run(SocketComponents {
1134            inspector: inspector.clone().into(),
1135            ctl_tx,
1136        })?;
1137
1138        state.incoming_segment(&segments[0].id(), (segments.len() as SeqNum).try_into()?)?;
1139
1140        tokio::time::sleep(cfg.expected_packet_latency * 2).await;
1141
1142        inspector
1143            .0
1144            .entry(1)
1145            .try_as_occupied()
1146            .ok_or(anyhow::anyhow!("frame 1 must be occupied"))?
1147            .get_mut()
1148            .add_segment(segments[1].clone())?;
1149
1150        state.incoming_segment(&segments[1].id(), (segments.len() as SeqNum).try_into()?)?;
1151
1152        tokio::time::sleep(cfg.expected_packet_latency * 2).await;
1153
1154        state.stop()?;
1155
1156        let ctl_msgs = tokio::time::timeout(Duration::from_millis(100), ctl_rx.collect::<Vec<_>>())
1157            .await
1158            .context("timeout receiving Control messages")?;
1159
1160        assert_eq!(2, ctl_msgs.len());
1161        assert_eq!(
1162            ctl_msgs[0],
1163            SessionMessage::Request(SegmentRequest::from_iter([(1, [0b01100000].into())]))
1164        );
1165
1166        assert_eq!(
1167            ctl_msgs[1],
1168            SessionMessage::Request(SegmentRequest::from_iter([(1, [0b00100000].into())]))
1169        );
1170
1171        Ok(())
1172    }
1173
1174    #[tokio::test]
1175    async fn ack_state_receiver_must_continue_requesting_missing_frames_and_acknowledge_once_complete()
1176    -> anyhow::Result<()> {
1177        let cfg = AcknowledgementStateConfig {
1178            mode: AcknowledgementMode::Partial,
1179            expected_packet_latency: Duration::from_millis(2),
1180            max_incoming_frame_retries: 3,
1181            acknowledgement_delay: Duration::from_millis(5),
1182            ..Default::default()
1183        };
1184
1185        let mut inspector = FrameInspector(FrameDashMap::with_capacity(10));
1186        let (ctl_tx, ctl_rx) = bounded_sink_channel::<SessionMessage<MTU>>(1024);
1187
1188        let segments = segment(hopr_types::crypto_random::random_bytes::<{ 2 * FRAME_SIZE }>(), MTU, 1)?;
1189
1190        // Segment 1
1191        inspector
1192            .0
1193            .entry(1)
1194            .try_as_vacant()
1195            .ok_or(anyhow::anyhow!("frame 1 must be vacant"))?
1196            .insert(FrameBuilder::from(segments[0].clone()));
1197
1198        let mut state = AcknowledgementState::<MTU>::new("test", cfg);
1199        state.run(SocketComponents {
1200            inspector: inspector.clone().into(),
1201            ctl_tx,
1202        })?;
1203
1204        // Segment 2
1205        state.incoming_segment(&segments[0].id(), (segments.len() as SeqNum).try_into()?)?;
1206
1207        tokio::time::sleep(cfg.expected_packet_latency * 2).await;
1208
1209        inspector
1210            .0
1211            .entry(1)
1212            .try_as_occupied()
1213            .ok_or(anyhow::anyhow!("frame 1 must be occupied"))?
1214            .get_mut()
1215            .add_segment(segments[1].clone())?;
1216
1217        state.incoming_segment(&segments[1].id(), (segments.len() as SeqNum).try_into()?)?;
1218
1219        tokio::time::sleep(cfg.expected_packet_latency * 2).await;
1220
1221        // Segment 3
1222        inspector
1223            .0
1224            .entry(1)
1225            .try_as_occupied()
1226            .ok_or(anyhow::anyhow!("frame 1 must be occupied"))?
1227            .get_mut()
1228            .add_segment(segments[2].clone())?;
1229
1230        state.incoming_segment(&segments[2].id(), (segments.len() as SeqNum).try_into()?)?;
1231        state.frame_complete(1)?;
1232
1233        tokio::time::sleep(cfg.acknowledgement_delay * 2).await;
1234
1235        state.stop()?;
1236
1237        let ctl_msgs = tokio::time::timeout(Duration::from_millis(100), ctl_rx.collect::<Vec<_>>())
1238            .await
1239            .context("timeout receiving Control messages")?;
1240
1241        assert_eq!(3, ctl_msgs.len());
1242        assert_eq!(
1243            ctl_msgs[0],
1244            SessionMessage::Request(SegmentRequest::from_iter([(1, [0b01100000].into())]))
1245        );
1246
1247        assert_eq!(
1248            ctl_msgs[1],
1249            SessionMessage::Request(SegmentRequest::from_iter([(1, [0b00100000].into())]))
1250        );
1251
1252        assert_eq!(ctl_msgs[2], SessionMessage::Acknowledge(vec![1].try_into()?));
1253
1254        Ok(())
1255    }
1256}
hopr_protocol_session/socket/ack_state.rs

hopr_protocol_session/socket/
ack_state.rs
