protocol/
sequence.rs

1//! Per-stream event sequence tracking (spec: Event channel with sequence
2//! numbers).
3//!
4//! Every stream numbers its events from 1 on each connection. On a lossless
5//! stream each event is one more than the last; anything else means the two
6//! sides disagree about the stream's state, so it goes out of sync until a
7//! resync names the last number the agent sent. A lossy stream (frame
8//! availability) only increases; the numbers it skips are counted as drops.
9
10use std::collections::HashMap;
11
12use crate::event::Stream;
13
14/// What to do with an event, given its stream and sequence number.
15#[derive(Clone, Copy, Debug, PartialEq, Eq)]
16pub enum SeqCheck {
17    /// Deliver it. `dropped` counts the numbers a lossy stream skipped before
18    /// it; always 0 on a lossless stream.
19    Deliver { dropped: u64 },
20    /// A lossless stream skipped or repeated a number and is now out of sync:
21    /// request a resync. The event itself is not delivered.
22    Gap { expected: u64, received: u64 },
23    /// Drop it: its stream awaits a resync, or a lossy stream went backwards.
24    Discard,
25}
26
27/// Sequence state of every stream seen on one connection.
28#[derive(Clone, Debug, Default)]
29pub struct StreamTracker {
30    streams: HashMap<Stream, StreamState>,
31}
32
33#[derive(Clone, Copy, Debug)]
34struct StreamState {
35    /// The last number accepted, or `None` while waiting for a resync.
36    last: Option<u64>,
37    /// Numbers a lossy stream skipped.
38    dropped: u64,
39}
40
41impl Default for StreamState {
42    fn default() -> StreamState {
43        StreamState {
44            last: Some(0),
45            dropped: 0,
46        }
47    }
48}
49
50impl StreamTracker {
51    pub fn new() -> StreamTracker {
52        StreamTracker::default()
53    }
54
55    /// Checks the next event of `stream`.
56    pub fn check(&mut self, stream: Stream, seq: u64) -> SeqCheck {
57        let state = self.streams.entry(stream).or_default();
58        let Some(last) = state.last else {
59            return SeqCheck::Discard;
60        };
61        if stream.is_lossy() {
62            if seq <= last {
63                return SeqCheck::Discard;
64            }
65            let dropped = seq - last - 1;
66            state.dropped += dropped;
67            state.last = Some(seq);
68            return SeqCheck::Deliver { dropped };
69        }
70        let expected = last.wrapping_add(1);
71        if seq == expected {
72            state.last = Some(seq);
73            SeqCheck::Deliver { dropped: 0 }
74        } else {
75            state.last = None;
76            SeqCheck::Gap {
77                expected,
78                received: seq,
79            }
80        }
81    }
82
83    /// The agent reports `stream` in sync again: `seq` is the last number it
84    /// sent on it.
85    pub fn resynced(&mut self, stream: Stream, seq: u64) {
86        self.streams.entry(stream).or_default().last = Some(seq);
87    }
88
89    /// Whether `stream` is in sync; a stream not seen yet is.
90    pub fn is_in_sync(&self, stream: Stream) -> bool {
91        self.streams.get(&stream).is_none_or(|s| s.last.is_some())
92    }
93
94    /// Numbers a lossy stream skipped so far.
95    pub fn dropped(&self, stream: Stream) -> u64 {
96        self.streams.get(&stream).map_or(0, |s| s.dropped)
97    }
98}
protocol/sequence.rs

protocol/
sequence.rs
