hopr_protocol_session/processing/
sequencer.rs

1//! This module defines the [`Sequencer`] stream adaptor.
2
3use std::{
4    collections::BinaryHeap,
5    future::Future,
6    pin::Pin,
7    task::{Context, Poll},
8    time::{Duration, Instant},
9};
10
11use futures_time::future::Timer;
12use tracing::instrument;
13
14use crate::{errors::SessionError, protocol::FrameId};
15
16/// Buffer entry pairing an item with when it entered the buffer.
17///
18/// Ordering delegates entirely to `item`, so the heap behaves exactly as before; `buffered_at`
19/// exists only to age the entry out.
20#[derive(Clone, Copy, Debug)]
21struct Buffered<T> {
22    item: T,
23    buffered_at: Instant,
24}
25
26impl<T: PartialEq> PartialEq for Buffered<T> {
27    fn eq(&self, other: &Self) -> bool {
28        self.item.eq(&other.item)
29    }
30}
31
32impl<T: Eq> Eq for Buffered<T> {}
33
34impl<T: PartialOrd> PartialOrd for Buffered<T> {
35    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
36        self.item.partial_cmp(&other.item)
37    }
38}
39
40impl<T: Ord> Ord for Buffered<T> {
41    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
42        self.item.cmp(&other.item)
43    }
44}
45
46impl<T: PartialOrd<FrameId>> PartialEq<FrameId> for Buffered<T> {
47    fn eq(&self, other: &FrameId) -> bool {
48        self.item.partial_cmp(other) == Some(std::cmp::Ordering::Equal)
49    }
50}
51
52impl<T: PartialOrd<FrameId>> PartialOrd<FrameId> for Buffered<T> {
53    fn partial_cmp(&self, other: &FrameId) -> Option<std::cmp::Ordering> {
54        self.item.partial_cmp(other)
55    }
56}
57
58/// Sequencer is an adaptor for streams, that yield elements that have a natural ordering and
59/// can be compared with [`FrameId`] and puts them in the correct sequence starting with
60/// `FrameId` equal to 1.
61///
62/// Sequencer internally maintains a `FrameId` to be yielded next, polls the underlying stream
63/// and yields elements only when they match the next `FrameId` to be yielded, incrementing the
64/// value on each yield.
65///
66/// The Sequencer takes to arguments: `max_wait` and `capacity`:
67///
68/// The `max_wait` indicates the maximum amount of time to wait for a certain `FrameId` to
69/// be yielded from the underlying stream.
70/// If this does not happen, the Segmenter yields an error,
71/// indicating that the given frame was discarded.
72///
73/// The `capacity` parameter sets the maximum number of buffered elements inside the Sequencer.
74/// If this value is reached, the Sequencer will stop polling the underlying stream, waiting for the
75/// next element to expire.
76///
77/// By definition, Sequencer is a fallible stream, yielding either `Ok(Item)`, `Err(`[`SessionError::FrameDiscarded`]`)`
78/// or `Ok(None)` when the underlying stream is closed and no more elements can be yielded.
79///
80/// Use [`SequencerExt`] methods to turn a stream into a sequenced stream.
81#[must_use = "streams do nothing unless polled"]
82#[pin_project::pin_project]
83pub struct Sequencer<S: futures::Stream> {
84    #[pin]
85    inner: S,
86    #[pin]
87    timer: futures_time::task::Sleep,
88    buffer: BinaryHeap<std::cmp::Reverse<Buffered<S::Item>>>,
89    next_id: FrameId,
90    last_emitted: Instant,
91    max_wait: Duration,
92    /// Anti-bufferbloat bound: items buffered longer than this are dropped, not emitted, so a
93    /// stall shows up as loss rather than a latency tail. `None` disables it.
94    max_item_age: Option<Duration>,
95    /// Head-of-line bound: abandon the frame due next once the sequence has advanced this far
96    /// past it, rather than holding everything for `max_wait`. `None` disables it.
97    max_frames_behind_gap: Option<usize>,
98    state: State,
99}
100
101impl<S> Sequencer<S>
102where
103    S: futures::Stream,
104    S::Item: Ord + PartialOrd<FrameId>,
105{
106    /// Creates a new instance, wrapping the given `inner` Segment sink.
107    ///
108    /// The `frame_size` value will be clamped into the `[C, (C - SessionMessage::SEGMENT_OVERHEAD) * SeqIndicator::MAX
109    /// + 1]` interval.
110    fn new(
111        inner: S,
112        max_wait: Duration,
113        capacity: usize,
114        max_item_age: Option<Duration>,
115        max_frames_behind_gap: Option<usize>,
116    ) -> Self {
117        assert!(capacity > 0, "capacity should be positive");
118        Self {
119            inner,
120            buffer: BinaryHeap::with_capacity(capacity),
121            timer: futures_time::task::sleep(max_wait.max(Duration::from_millis(1)).into()),
122            next_id: 1,
123            last_emitted: Instant::now(),
124            max_wait,
125            max_item_age: max_item_age.filter(|age| !age.is_zero()),
126            // Zero would abandon the frame due next before anything had arrived to justify it,
127            // turning every momentary gap into loss. One later frame is the strictest evidence
128            // that still *is* evidence.
129            max_frames_behind_gap: max_frames_behind_gap.map(|n| n.max(1)),
130            state: State::Polling,
131        }
132    }
133}
134
135#[derive(Clone, Copy, Debug, PartialEq, Eq)]
136enum State {
137    Polling,
138    BufferUpdated,
139    Done,
140}
141
142impl<S> futures::Stream for Sequencer<S>
143where
144    S: futures::Stream,
145    S::Item: Ord + PartialOrd<FrameId>,
146{
147    type Item = Result<S::Item, SessionError>;
148
149    #[instrument(name = "Sequencer::poll_next", level = "trace", skip(self, cx), fields(next_frame_id = self.next_id, state = ?self.state))]
150    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
151        let mut this = self.project();
152        if *this.next_id == 0 {
153            tracing::debug!("end of frame sequence reached");
154            return Poll::Ready(None);
155        }
156
157        loop {
158            match *this.state {
159                State::Polling => {
160                    if this.buffer.len() < this.buffer.capacity() {
161                        // We still have capacity available, poll the underlying stream
162                        let stream_poll = this.inner.as_mut().poll_next(cx);
163
164                        // Only poll timer if there's something in the buffer
165                        let timer_poll = if !this.buffer.is_empty() {
166                            let poll = this.timer.as_mut().poll(cx);
167                            if poll.is_ready() {
168                                this.timer.as_mut().reset_timer();
169                            }
170                            poll
171                        } else {
172                            Poll::Pending
173                        };
174
175                        match (stream_poll, timer_poll) {
176                            (Poll::Pending, Poll::Pending) => {
177                                tracing::trace!("pending");
178                                *this.state = State::Polling;
179                                return Poll::Pending;
180                            }
181                            (Poll::Ready(Some(item)), _) => {
182                                // We have to reset the last emitted timestamp if
183                                // the buffer was empty until now
184                                if this.buffer.is_empty() {
185                                    *this.last_emitted = Instant::now();
186                                }
187
188                                if item.lt(this.next_id) {
189                                    // Do not accept older frame ids
190                                    tracing::error!("old item");
191                                    *this.state = State::Polling;
192                                } else {
193                                    // Push new item to the buffer
194                                    tracing::trace!("new item");
195                                    this.buffer.push(std::cmp::Reverse(Buffered {
196                                        item,
197                                        buffered_at: Instant::now(),
198                                    }));
199                                    *this.state = State::BufferUpdated;
200                                }
201                            }
202                            (Poll::Ready(None), _) => {
203                                tracing::trace!(len = this.buffer.len(), "stream is done");
204                                *this.state = State::Done
205                            }
206                            (_, Poll::Ready(_)) => {
207                                // Simulate buffer update when the timer elapses
208                                tracing::trace!("timer elapsed");
209                                *this.state = State::BufferUpdated;
210                            }
211                        }
212                    } else {
213                        // Simulate buffer update when at capacity
214                        tracing::warn!("sequencer buffer is full");
215                        *this.state = State::BufferUpdated;
216                    }
217                }
218                State::BufferUpdated => {
219                    // The buffer has been updated, check if we can yield something
220                    if let Some(next) = this.buffer.peek().map(|item| &item.0) {
221                        if next.eq(this.next_id) {
222                            let stale = this
223                                .max_item_age
224                                .is_some_and(|max_age| next.buffered_at.elapsed() >= max_age);
225
226                            *this.next_id = this.next_id.wrapping_add(1);
227                            *this.last_emitted = Instant::now();
228                            *this.state = State::BufferUpdated;
229
230                            // Anti-bufferbloat: the frame is the one due next, but it has been held
231                            // so long that delivering it would only add latency. Report it as
232                            // discarded — the same signal the consumer already handles for a lost
233                            // frame — instead of emitting it late.
234                            if stale {
235                                let discarded = this.next_id.wrapping_sub(1);
236                                this.buffer.pop();
237                                tracing::trace!(discarded, "discard frame that exceeded max age");
238                                return Poll::Ready(Some(Err(SessionError::FrameDiscarded(discarded))));
239                            }
240
241                            tracing::trace!("emit next frame");
242
243                            return Poll::Ready(this.buffer.pop().map(|item| Ok(item.0.item)));
244                        } else if this.last_emitted.elapsed() >= *this.max_wait
245                            || this.buffer.len() == this.buffer.capacity()
246                            // The sequence has moved far enough past the gap to conclude the
247                            // missing frame was lost rather than reordered. Waiting out `max_wait`
248                            // cannot change that verdict, and on a session with no retransmission
249                            // nothing else can either.
250                            || this.max_frames_behind_gap.is_some_and(|n| {
251                                // Two readings of the same evidence, because each is blind where
252                                // the other sees. The count catches a run of frames arriving
253                                // behind the gap. The distance catches the case that actually
254                                // dominates under loss: the frames in between never completed, so
255                                // they never reach the sequencer and the buffer stays nearly
256                                // empty however far the sender has moved on.
257                                this.buffer.len() >= n
258                                    // Saturating, not `as`: the bound is configuration, and a value past `FrameId`'s
259                                // range would wrap into a small threshold rather than a large one.
260                                || next.gt(&this.next_id.saturating_add(
261                                    FrameId::try_from(n).unwrap_or(FrameId::MAX).saturating_sub(1),
262                                ))
263                            })
264                        {
265                            let discarded = *this.next_id;
266                            *this.next_id = this.next_id.wrapping_add(1);
267                            // `last_emitted` is intentionally NOT reset here: it is only reset
268                            // when an actual frame is emitted. Resetting it per discarded id would
269                            // drain a contiguous gap of K missing frames at 1 frame per `max_wait`
270                            // (a K x max_wait delivery stall of frames already sitting in the
271                            // buffer), instead of flushing the whole gap once `max_wait` elapses.
272                            *this.state = State::BufferUpdated;
273
274                            tracing::trace!(discarded, "discard frame");
275
276                            return Poll::Ready(Some(Err(SessionError::FrameDiscarded(discarded))));
277                        }
278                    } else {
279                        tracing::trace!("buffer is empty");
280                    }
281
282                    // Nothing to yield, keep on polling
283                    *this.state = State::Polling;
284                }
285                State::Done => {
286                    // The underlying stream is done, drain what we have in the internal buffer
287                    return if let Some(next) = this.buffer.peek().map(|item| &item.0) {
288                        if next.lt(this.next_id) {
289                            tracing::error!("old item");
290                            this.buffer.pop();
291                            continue;
292                        } else if next.eq(this.next_id) {
293                            *this.next_id = this.next_id.wrapping_add(1);
294                            tracing::trace!("emit next frame when done");
295
296                            Poll::Ready(this.buffer.pop().map(|item| Ok(item.0.item)))
297                        } else {
298                            let discarded = *this.next_id;
299                            *this.next_id = this.next_id.wrapping_add(1);
300                            tracing::trace!(discarded, "discard frame when done");
301
302                            Poll::Ready(Some(Err(SessionError::FrameDiscarded(discarded))))
303                        }
304                    } else {
305                        tracing::trace!("buffer is empty and done");
306                        Poll::Ready(None)
307                    };
308                }
309            }
310        }
311    }
312}
313
314/// How a [`Sequencer`] decides to stop waiting for a frame that has not arrived.
315///
316/// Grouped rather than passed positionally: the two stopping rules below answer different
317/// questions, and a bare list of `Duration`/`Option<usize>` arguments at the call site gives no
318/// hint which is which.
319#[derive(Clone, Copy, Debug)]
320pub struct SequencerConfig {
321    /// Longest the frame due next is waited for before it is abandoned.
322    pub max_wait: Duration,
323    /// Maximum number of buffered items.
324    pub capacity: usize,
325    /// Discards items buffered longer than this instead of emitting them late; `None` disables.
326    pub max_item_age: Option<Duration>,
327    /// Abandons the frame due next once this many later frames are already waiting behind it.
328    ///
329    /// `None` waits out [`Self::max_wait`] regardless of how much has piled up, which is correct
330    /// only when the missing frame can still be recovered. On a session without retransmission it
331    /// cannot: the wait is for something that is never coming, and everything already received is
332    /// held behind it for the full duration.
333    ///
334    /// Counting later frames rather than watching a clock makes the decision on evidence. One or
335    /// two frames arriving ahead is ordinary reordering across paths of differing latency; a queue
336    /// building up behind a gap is a frame that was lost.
337    pub max_frames_behind_gap: Option<usize>,
338}
339
340/// Stream extensions methods for item sequencing.
341pub trait SequencerExt: futures::Stream {
342    /// Attaches a [`Sequencer`] to the underlying stream, given the item `timeout` and `capacity`
343    /// of items.
344    fn sequencer(self, timeout: Duration, capacity: usize) -> Sequencer<Self>
345    where
346        Self::Item: Ord + PartialOrd<FrameId>,
347        Self: Sized,
348    {
349        Sequencer::new(self, timeout, capacity, None, None)
350    }
351
352    /// As [`SequencerExt::sequencer`], but discards items buffered longer than `max_item_age`
353    /// instead of emitting them late.
354    fn sequencer_with_max_age(
355        self,
356        timeout: Duration,
357        capacity: usize,
358        max_item_age: Option<Duration>,
359    ) -> Sequencer<Self>
360    where
361        Self::Item: Ord + PartialOrd<FrameId>,
362        Self: Sized,
363    {
364        Sequencer::new(self, timeout, capacity, max_item_age, None)
365    }
366
367    /// As [`SequencerExt::sequencer`], with every stopping rule stated explicitly.
368    fn sequencer_with(self, cfg: SequencerConfig) -> Sequencer<Self>
369    where
370        Self::Item: Ord + PartialOrd<FrameId>,
371        Self: Sized,
372    {
373        Sequencer::new(
374            self,
375            cfg.max_wait,
376            cfg.capacity,
377            cfg.max_item_age,
378            cfg.max_frames_behind_gap,
379        )
380    }
381}
382
383impl<T: ?Sized> SequencerExt for T where T: futures::Stream {}
384
385#[cfg(test)]
386mod tests {
387    use futures::{SinkExt, StreamExt, TryStreamExt, pin_mut};
388    use futures_time::future::FutureExt;
389
390    use super::*;
391
392    #[test_log::test(tokio::test)]
393    async fn sequencer_should_return_entries_in_order() -> anyhow::Result<()> {
394        let mut expected = vec![4u32, 1, 5, 7, 8, 6, 2, 3];
395
396        let actual: Vec<u32> = futures::stream::iter(expected.clone())
397            .sequencer(Duration::from_secs(5), 4096)
398            .try_collect()
399            .timeout(futures_time::time::Duration::from_secs(5))
400            .await??;
401
402        expected.sort();
403        assert_eq!(expected, actual);
404
405        Ok(())
406    }
407
408    #[test_log::test(tokio::test)]
409    async fn sequencer_should_discard_entries_that_exceeded_the_max_age() -> anyhow::Result<()> {
410        let (seq_sink, seq_stream) = futures::channel::mpsc::unbounded();
411
412        // `max_wait` is long, so nothing here is discarded for being *missing* — only for being stale.
413        let seq_stream =
414            seq_stream.sequencer_with_max_age(Duration::from_secs(30), 4096, Some(Duration::from_millis(100)));
415
416        pin_mut!(seq_sink);
417        pin_mut!(seq_stream);
418
419        // Frame 2 arrives first and waits in the buffer for the missing frame 1 — a transport stall.
420        seq_sink.send(2u32).await?;
421
422        // Drive the sequencer so frame 2 actually lands in its buffer; until it is polled the item
423        // only sits in the channel and its buffered-at clock has not started.
424        assert!(
425            seq_stream
426                .try_next()
427                .timeout(futures_time::time::Duration::from_millis(50))
428                .await
429                .is_err(),
430            "nothing is emitted while frame 1 is missing"
431        );
432
433        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
434
435        // Frame 1 arrives fresh and is delivered; frame 2 is now 250 ms stale and must be
436        // reported as discarded rather than handed over a quarter of a second late.
437        seq_sink.send(1u32).await?;
438
439        assert_eq!(Some(1), seq_stream.try_next().await?, "the fresh frame is delivered");
440        assert!(
441            matches!(seq_stream.try_next().await, Err(SessionError::FrameDiscarded(2))),
442            "the stale frame must be discarded, not delivered late"
443        );
444
445        Ok(())
446    }
447
448    #[test_log::test(tokio::test)]
449    async fn sequencer_should_deliver_entries_within_the_max_age() -> anyhow::Result<()> {
450        let (seq_sink, seq_stream) = futures::channel::mpsc::unbounded();
451
452        let seq_stream =
453            seq_stream.sequencer_with_max_age(Duration::from_secs(30), 4096, Some(Duration::from_secs(30)));
454
455        pin_mut!(seq_sink);
456        pin_mut!(seq_stream);
457
458        // Same out-of-order arrival and the same buffering delay, but comfortably inside the
459        // bound: nothing is dropped.
460        seq_sink.send(2u32).await?;
461        assert!(
462            seq_stream
463                .try_next()
464                .timeout(futures_time::time::Duration::from_millis(50))
465                .await
466                .is_err()
467        );
468        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
469        seq_sink.send(1u32).await?;
470
471        assert_eq!(Some(1), seq_stream.try_next().await?);
472        assert_eq!(Some(2), seq_stream.try_next().await?);
473
474        Ok(())
475    }
476
477    #[test_log::test(tokio::test)]
478    async fn sequencer_should_not_allow_emitted_entries() -> anyhow::Result<()> {
479        let (seq_sink, seq_stream) = futures::channel::mpsc::unbounded();
480
481        let seq_stream = seq_stream.sequencer(Duration::from_secs(1), 4096);
482
483        pin_mut!(seq_sink);
484        pin_mut!(seq_stream);
485
486        seq_sink.send(1u32).await?;
487        assert_eq!(Some(1), seq_stream.try_next().await?);
488
489        seq_sink.send(2u32).await?;
490        assert_eq!(Some(2), seq_stream.try_next().await?);
491
492        seq_sink.send(2u32).await?;
493        seq_sink.send(1u32).await?;
494
495        seq_sink.send(3u32).await?;
496        assert_eq!(Some(3), seq_stream.try_next().await?);
497
498        Ok(())
499    }
500
501    /// Frames waiting behind a gap must not be held for `max_wait`.
502    ///
503    /// This is the head-of-line stall, measured on a live cluster: with `max_wait` at 3 s a
504    /// session returning 98.5 % of its bytes over the wire delivered 0.60 % of them to the
505    /// application, and the application-side inter-arrival median sat exactly on the timeout. On a
506    /// session with no retransmission the missing frame is never coming, so every second of that
507    /// wait is spent on an outcome that cannot change while the frames already received are held.
508    #[test_log::test(tokio::test)]
509    async fn sequencer_should_abandon_a_gap_once_enough_later_frames_are_waiting() -> anyhow::Result<()> {
510        // Far longer than the test should take, so any reliance on the timer trips the bound below
511        // rather than becoming a matter of tuning margins.
512        let max_wait = Duration::from_secs(5);
513        let (mut seq_sink, seq_stream) = futures::channel::mpsc::unbounded();
514
515        // Frame 1 never arrives.
516        for v in [2u32, 3, 4] {
517            seq_sink.feed(v).await?;
518        }
519        seq_sink.flush().await?;
520
521        let seq_stream = seq_stream.sequencer_with(SequencerConfig {
522            max_wait,
523            capacity: 4096,
524            max_item_age: None,
525            max_frames_behind_gap: Some(2),
526        });
527        pin_mut!(seq_stream);
528
529        // The bound is the assertion: without the rule the sequencer holds everything for the full
530        // `max_wait`, so a regression trips the timeout instead of blocking the suite for 5 s.
531        let released: Vec<u32> = tokio::time::timeout(max_wait / 2, async {
532            assert!(
533                matches!(seq_stream.try_next().await, Err(SessionError::FrameDiscarded(1))),
534                "the gap must be reported as loss, the signal the consumer already handles"
535            );
536            seq_stream.by_ref().take(3).try_collect().await
537        })
538        .await
539        .map_err(|_| anyhow::anyhow!("frames already received waited on a frame that is never coming"))??;
540
541        assert_eq!(vec![2, 3, 4], released, "everything behind the gap must follow it out");
542
543        // Held open deliberately: closing the sink would drain the buffer through `State::Done`,
544        // which has its own emit path and would pass this test without the rule under test.
545        drop(seq_sink);
546        Ok(())
547    }
548
549    /// Under real loss the frames between the gap and the newest arrival mostly never complete, so
550    /// they never reach the sequencer at all and the buffer stays nearly empty. A rule counting
551    /// buffered frames then never fires and the timeout takes over — which is exactly what a
552    /// cluster showed: at an identical setting the scenario delivered 95–97 % on some runs and
553    /// 0.5 % on others, the failing ones with an application-side inter-arrival median sitting
554    /// back on the 3 s timeout.
555    ///
556    /// What is available regardless is how far the sequence has advanced past the gap. One frame
557    /// arriving at id 40 says as much about frame 1 as forty buffered frames would.
558    #[test_log::test(tokio::test)]
559    async fn sequencer_should_abandon_a_gap_when_the_sequence_has_advanced_past_it() -> anyhow::Result<()> {
560        let max_wait = Duration::from_secs(5);
561        let (mut seq_sink, seq_stream) = futures::channel::mpsc::unbounded();
562
563        // Frame 1 is missing and frames 2..=39 never completed, so only one frame is buffered --
564        // far below any count-based threshold, yet the sequence has clearly moved on.
565        seq_sink.feed(40u32).await?;
566        seq_sink.flush().await?;
567
568        let seq_stream = seq_stream.sequencer_with(SequencerConfig {
569            max_wait,
570            capacity: 4096,
571            max_item_age: None,
572            max_frames_behind_gap: Some(4),
573        });
574        pin_mut!(seq_stream);
575
576        // Only the frames that the sequence has genuinely left behind. The last `n - 1` before the
577        // newest arrival are still inside the reordering window -- the sequence has not advanced
578        // far enough past *them* to call them lost -- so they keep the timeout, which is the
579        // conservative half of the same rule.
580        let now = Instant::now();
581        for expected_gap in 1..=36u32 {
582            assert!(
583                matches!(
584                    seq_stream.try_next().await,
585                    Err(SessionError::FrameDiscarded(id)) if id == expected_gap
586                ),
587                "frame {expected_gap} is behind the advanced sequence and must be given up on"
588            );
589        }
590        assert!(
591            now.elapsed() < max_wait / 2,
592            "a single frame far ahead is evidence enough, with no frames buffered behind the gap to count; took {:?}",
593            now.elapsed()
594        );
595
596        drop(seq_sink);
597        Ok(())
598    }
599
600    /// The inverse, so the rule cannot become "never wait". Below the threshold the sequencer must
601    /// still hold the gap open — one frame arriving ahead is ordinary reordering across paths of
602    /// differing latency, not a loss.
603    #[test_log::test(tokio::test)]
604    async fn sequencer_should_keep_waiting_while_fewer_frames_are_behind_the_gap() -> anyhow::Result<()> {
605        let max_wait = Duration::from_millis(300);
606        let (mut seq_sink, seq_stream) = futures::channel::mpsc::unbounded();
607
608        // Frame 1 is missing and only one frame is waiting behind it, under the threshold of 3.
609        seq_sink.feed(2u32).await?;
610        seq_sink.flush().await?;
611
612        let seq_stream = seq_stream.sequencer_with(SequencerConfig {
613            max_wait,
614            capacity: 4096,
615            max_item_age: None,
616            max_frames_behind_gap: Some(3),
617        });
618        pin_mut!(seq_stream);
619
620        let now = Instant::now();
621        assert!(matches!(
622            seq_stream.try_next().await,
623            Err(SessionError::FrameDiscarded(1))
624        ));
625        assert!(
626            now.elapsed() >= max_wait,
627            "under the threshold the timeout still governs; took {:?}",
628            now.elapsed()
629        );
630
631        drop(seq_sink);
632        Ok(())
633    }
634
635    /// A session that *can* recover a missing frame must be unaffected: `None` keeps the timeout
636    /// as the only rule, however much piles up behind the gap.
637    #[test_log::test(tokio::test)]
638    async fn sequencer_without_the_gap_bound_should_still_wait_for_the_timeout() -> anyhow::Result<()> {
639        let max_wait = Duration::from_millis(300);
640        let (mut seq_sink, seq_stream) = futures::channel::mpsc::unbounded();
641
642        for v in [2u32, 3, 4, 5, 6] {
643            seq_sink.feed(v).await?;
644        }
645        seq_sink.flush().await?;
646
647        let seq_stream = seq_stream.sequencer_with(SequencerConfig {
648            max_wait,
649            capacity: 4096,
650            max_item_age: None,
651            max_frames_behind_gap: None,
652        });
653        pin_mut!(seq_stream);
654
655        let now = Instant::now();
656        assert!(matches!(
657            seq_stream.try_next().await,
658            Err(SessionError::FrameDiscarded(1))
659        ));
660        assert!(
661            now.elapsed() >= max_wait,
662            "with no gap bound the timeout is the only rule; took {:?}",
663            now.elapsed()
664        );
665
666        drop(seq_sink);
667        Ok(())
668    }
669
670    #[test_log::test(tokio::test)]
671    async fn sequencer_should_discard_entry_on_timeout() -> anyhow::Result<()> {
672        let timeout = Duration::from_millis(25);
673        let (mut seq_sink, seq_stream) = futures::channel::mpsc::unbounded();
674
675        let input = vec![2u32, 1, 4, 5, 8, 7, 9, 11, 10];
676
677        let input_clone = input.clone();
678        let jh = hopr_utils::runtime::prelude::spawn(async move {
679            for v in input_clone {
680                seq_sink
681                    .feed(v)
682                    .delay(futures_time::time::Duration::from_millis(5))
683                    .await?;
684            }
685            seq_sink.flush().await?;
686            seq_sink.close().await
687        });
688
689        let seq_stream = seq_stream.sequencer(timeout, 4096);
690
691        pin_mut!(seq_stream);
692
693        assert_eq!(Some(1), seq_stream.try_next().await?);
694        assert_eq!(Some(2), seq_stream.try_next().await?);
695
696        let now = Instant::now();
697        assert!(matches!(
698            seq_stream.try_next().await,
699            Err(SessionError::FrameDiscarded(3))
700        ));
701        assert!(now.elapsed() >= timeout);
702
703        assert_eq!(Some(4), seq_stream.try_next().await?);
704        assert_eq!(Some(5), seq_stream.try_next().await?);
705
706        assert!(matches!(
707            seq_stream.try_next().await,
708            Err(SessionError::FrameDiscarded(6))
709        ));
710
711        assert_eq!(Some(7), seq_stream.try_next().await?);
712        assert_eq!(Some(8), seq_stream.try_next().await?);
713        assert_eq!(Some(9), seq_stream.try_next().await?);
714        assert_eq!(Some(10), seq_stream.try_next().await?);
715        assert_eq!(Some(11), seq_stream.try_next().await?);
716
717        assert_eq!(None, seq_stream.try_next().await?);
718
719        let _ = jh.await?;
720        Ok(())
721    }
722
723    #[test_log::test(tokio::test)]
724    async fn sequencer_should_discard_entry_close() -> anyhow::Result<()> {
725        let (seq_sink, seq_stream) = futures::channel::mpsc::unbounded();
726
727        let input = vec![2u32, 1, 3, 5, 4, 8, 11];
728
729        hopr_utils::runtime::prelude::spawn(futures::stream::iter(input.clone()).map(Ok).forward(seq_sink)).await??;
730
731        let seq_stream = seq_stream.sequencer(Duration::from_millis(25), 4096);
732
733        pin_mut!(seq_stream);
734
735        assert_eq!(Some(1), seq_stream.try_next().await?);
736        assert_eq!(Some(2), seq_stream.try_next().await?);
737        assert_eq!(Some(3), seq_stream.try_next().await?);
738        assert_eq!(Some(4), seq_stream.try_next().await?);
739        assert_eq!(Some(5), seq_stream.try_next().await?);
740        assert!(matches!(
741            seq_stream.try_next().await,
742            Err(SessionError::FrameDiscarded(6))
743        ));
744        assert!(matches!(
745            seq_stream.try_next().await,
746            Err(SessionError::FrameDiscarded(7))
747        ));
748        assert_eq!(Some(8), seq_stream.try_next().await?);
749        assert!(matches!(
750            seq_stream.try_next().await,
751            Err(SessionError::FrameDiscarded(9))
752        ));
753        assert!(matches!(
754            seq_stream.try_next().await,
755            Err(SessionError::FrameDiscarded(10))
756        ));
757        assert_eq!(Some(11), seq_stream.try_next().await?);
758        assert_eq!(None, seq_stream.try_next().await?);
759
760        Ok(())
761    }
762
763    #[test_log::test(tokio::test)]
764    async fn sequencer_should_discard_entry_when_inner_stream_pending() -> anyhow::Result<()> {
765        let sent = vec![4u32, 1, 7, 8, 6, 2, 3];
766        let (tx, rx) = futures::channel::mpsc::unbounded();
767
768        pin_mut!(tx);
769        tx.send_all(&mut futures::stream::iter(sent.clone()).map(Ok)).await?;
770
771        let rx = rx.sequencer(Duration::from_millis(10), 4096);
772        pin_mut!(rx);
773
774        assert!(matches!(rx.next().await, Some(Ok(1))));
775        assert!(matches!(rx.next().await, Some(Ok(2))));
776        assert!(matches!(rx.next().await, Some(Ok(3))));
777        assert!(matches!(rx.next().await, Some(Ok(4))));
778        assert!(matches!(rx.next().await, Some(Err(SessionError::FrameDiscarded(5)))));
779        assert!(matches!(rx.next().await, Some(Ok(6))));
780        assert!(matches!(rx.next().await, Some(Ok(7))));
781        assert!(matches!(rx.next().await, Some(Ok(8))));
782
783        Ok(())
784    }
785
786    #[test_log::test(tokio::test)]
787    async fn sequencer_should_discard_entry_when_capacity_is_reached() -> anyhow::Result<()> {
788        let sent = vec![4u32, 5, 7, 8, 2, 6, 3];
789        let (tx, rx) = futures::channel::mpsc::unbounded();
790
791        pin_mut!(tx);
792        tx.send_all(&mut futures::stream::iter(sent.clone()).map(Ok)).await?;
793
794        let rx = rx.sequencer(Duration::from_millis(10), 4);
795        pin_mut!(rx);
796
797        assert!(matches!(rx.next().await, Some(Err(SessionError::FrameDiscarded(1)))));
798        assert!(matches!(rx.next().await, Some(Err(SessionError::FrameDiscarded(2)))));
799        assert!(matches!(rx.next().await, Some(Err(SessionError::FrameDiscarded(3)))));
800        assert!(matches!(rx.next().await, Some(Ok(4))));
801        assert!(matches!(rx.next().await, Some(Ok(5))));
802        assert!(matches!(rx.next().await, Some(Ok(6))));
803        assert!(matches!(rx.next().await, Some(Ok(7))));
804        assert!(matches!(rx.next().await, Some(Ok(8))));
805
806        Ok(())
807    }
808
809    #[test_log::test(tokio::test)]
810    async fn sequencer_should_drain_contiguous_gap_within_single_timeout_window() -> anyhow::Result<()> {
811        let timeout = Duration::from_millis(50);
812        let (tx, rx) = futures::channel::mpsc::unbounded();
813
814        pin_mut!(tx);
815        tx.send_all(&mut futures::stream::iter([1u32, 2, 10, 11, 12]).map(Ok))
816            .await?;
817
818        let rx = rx.sequencer(timeout, 4096);
819        pin_mut!(rx);
820
821        assert_eq!(Some(1), rx.try_next().await?);
822        assert_eq!(Some(2), rx.try_next().await?);
823
824        let now = Instant::now();
825        for expected in 3u32..=9 {
826            assert!(matches!(
827                rx.next().await,
828                Some(Err(SessionError::FrameDiscarded(id))) if id == expected
829            ));
830        }
831        assert_eq!(Some(10), rx.try_next().await?);
832        assert_eq!(Some(11), rx.try_next().await?);
833        assert_eq!(Some(12), rx.try_next().await?);
834
835        // The 7-frame gap must be flushed after one timeout window,
836        // not at a rate of one frame per window.
837        assert!(
838            now.elapsed() < 3 * timeout,
839            "gap drain took {:?}, expected well under {:?}",
840            now.elapsed(),
841            7 * timeout
842        );
843
844        Ok(())
845    }
846
847    #[test_log::test(tokio::test)]
848    async fn sequencer_must_terminate_on_last_frame_id() -> anyhow::Result<()> {
849        let (tx, rx) = futures::channel::mpsc::unbounded();
850
851        pin_mut!(tx);
852        tx.send_all(&mut futures::stream::iter([FrameId::MAX - 1, FrameId::MAX, 1, 2]).map(Ok))
853            .await?;
854
855        let mut rx = rx.sequencer(Duration::from_millis(10), 1024);
856        rx.next_id = FrameId::MAX - 1;
857        pin_mut!(rx);
858
859        const LAST_ID: FrameId = FrameId::MAX - 1;
860        assert!(matches!(rx.next().await, Some(Ok(LAST_ID))));
861        assert!(matches!(rx.next().await, Some(Ok(FrameId::MAX))));
862        assert!(rx.next().await.is_none());
863
864        Ok(())
865    }
866
867    #[test_log::test(tokio::test(flavor = "multi_thread"))]
868    async fn sequencer_must_not_discard_frames_when_buffer_was_empty_after_timeout() -> anyhow::Result<()> {
869        let (tx, rx) = futures::channel::mpsc::unbounded();
870
871        let jh = tokio::task::spawn(async move {
872            tokio::time::sleep(Duration::from_millis(2)).await;
873            pin_mut!(tx);
874            tx.send_all(&mut futures::stream::iter([3, 1, 2, 4]).map(Ok)).await?;
875
876            tokio::time::sleep(Duration::from_millis(150)).await;
877
878            tx.send_all(&mut futures::stream::iter([6, 5, 7]).map(Ok)).await?;
879
880            anyhow::Ok(())
881        });
882
883        let chunks = rx
884            .sequencer(Duration::from_millis(50), 1024)
885            .try_ready_chunks(10)
886            .try_collect::<Vec<Vec<_>>>()
887            .await?;
888
889        assert_eq!(chunks, vec![vec![1, 2, 3, 4], vec![5, 6, 7]]);
890        jh.await??;
891
892        Ok(())
893    }
894}
hopr_protocol_session/processing/sequencer.rs

hopr_protocol_session/processing/
sequencer.rs
