srt_runtime/arq/
receiver.rs

1//! ARQ receiver-side reliability state — `draft-sharabayko-srt-01` §4.8.1
2//! (Full/Light ACK generation), §4.8.2 (loss detection + NAK), §4.10 (RTT
3//! measurement via the ACK/ACKACK round trip). Curated rules:
4//! `specs/rules/srt-arq.md`.
5//!
6//! Sans-IO: [`Receiver`] never reads a wall clock — [`Receiver::feed_data`]
7//! and [`Receiver::tick`] take a caller-supplied `now: core::time::Duration`
8//! (elapsed time since a fixed epoch the caller owns).
9//!
10//! # Delivery model
11//! [`FeedOutcome::delivered`] lists the sequence numbers that became
12//! cumulatively in-order-deliverable as a result of one `feed_data` call —
13//! i.e. the ARQ engine's view of "no longer missing" (rule 8's ack point),
14//! not a TSBPD-timed playout (that delay is `srt-tsbpd.md` scope, a
15//! separate follow-up).
16//!
17//! # Non-goals
18//! TLPKTDROP fake-ACK skip handling (rule 13) is not modeled — see the
19//! `arq` module doc.
20
21use alloc::collections::{BTreeMap, BTreeSet};
22use alloc::vec::Vec;
23use core::time::Duration;
24
25use crate::packet::nak::build_loss_list;
26use crate::packet::{AckAckPacket, AckCif, AckPacket, ControlPacket, LossListEntry, NakPacket};
27
28use super::rtt::RttEstimator;
29use super::{FULL_ACK_PERIOD, LIGHT_ACK_THRESHOLD, duration_to_wire_us, nak_interval, seq};
30
31/// A bound on how many individual sequence numbers one [`Receiver::feed_data`]
32/// call will enumerate into the loss list for a single newly-detected gap.
33/// Not a `specs/rules/srt-arq.md` rule — a safety cap against a corrupt or
34/// adversarial sequence-number jump causing unbounded work (mirrors the
35/// NAK-side cap in `arq::sender::expand_loss_entry`).
36const MAX_GAP_EXPANSION: u32 = 1 << 16;
37
38/// Outcome of one [`Receiver::feed_data`] call.
39#[derive(Debug, Clone, PartialEq, Eq, Default)]
40#[non_exhaustive]
41pub struct FeedOutcome {
42    /// Sequence numbers that became cumulatively in-order-deliverable as a
43    /// result of this packet's arrival — includes `seq` itself when it was
44    /// the next-expected packet, plus any previously-buffered out-of-order
45    /// packets it consequently unblocked.
46    pub delivered: Vec<u32>,
47    /// An immediate NAK to send, if this packet's arrival revealed a new
48    /// gap (`specs/rules/srt-arq.md` rules 4, 14).
49    pub nak: Option<Vec<u8>>,
50}
51
52/// ARQ receiver-side state (`draft-sharabayko-srt-01` §4.8.1/§4.8.2/§4.10).
53#[derive(Debug)]
54pub struct Receiver {
55    dest_socket_id: u32,
56    /// Cumulative ack point: every seq strictly before this has been
57    /// delivered in order (rule 8).
58    next_expected: u32,
59    /// Received but not yet in-order-deliverable (a gap remains below it).
60    out_of_order: BTreeSet<u32>,
61    /// Sequence numbers currently believed lost (rules 4, 14, 21).
62    loss_list: BTreeSet<u32>,
63    /// The highest sequence number ever received, to detect newly-opened
64    /// gaps (bookkeeping, not itself a spec-named field).
65    highest_received: Option<u32>,
66    /// Packets received since the last ACK of either kind (rule 12).
67    packets_since_ack: u32,
68    /// Absolute time the last Full ACK was sent (rule 11).
69    last_full_ack_at: Duration,
70    /// Absolute time the last periodic NAK was sent (rule 22).
71    last_nak_at: Duration,
72    /// Next Full ACK's Acknowledgement Number ("starting from 1", §3.2.4).
73    next_ack_number: u32,
74    /// Outstanding Full ACKs awaiting their ACKACK, send-time keyed by
75    /// Acknowledgement Number (rules 24, 26-28).
76    outstanding_acks: BTreeMap<u32, Duration>,
77    rtt: RttEstimator,
78}
79
80impl Receiver {
81    /// A fresh receiver expecting `initial_seq` first (the peer's ISN),
82    /// addressing `dest_socket_id` (the peer's SRT Socket ID, §3).
83    pub fn new(dest_socket_id: u32, initial_seq: u32) -> Self {
84        Receiver {
85            dest_socket_id,
86            next_expected: initial_seq,
87            out_of_order: BTreeSet::new(),
88            loss_list: BTreeSet::new(),
89            highest_received: None,
90            packets_since_ack: 0,
91            last_full_ack_at: Duration::ZERO,
92            last_nak_at: Duration::ZERO,
93            next_ack_number: 1,
94            outstanding_acks: BTreeMap::new(),
95            rtt: RttEstimator::new(),
96        }
97    }
98
99    /// The cumulative ack point — every seq strictly before this has been
100    /// delivered in order (rule 8's ACK `n + 1` semantics).
101    pub fn ack_point(&self) -> u32 {
102        self.next_expected
103    }
104
105    /// The current RTT estimate (rules 26-31).
106    pub fn rtt(&self) -> Duration {
107        self.rtt.rtt()
108    }
109
110    /// The current RTTVar estimate.
111    pub fn rtt_var(&self) -> Duration {
112        self.rtt.rtt_var()
113    }
114
115    /// Number of sequence numbers currently believed lost.
116    pub fn loss_list_len(&self) -> usize {
117        self.loss_list.len()
118    }
119
120    /// Process one arriving data packet's sequence number.
121    pub fn feed_data(&mut self, seq_number: u32, now: Duration) -> FeedOutcome {
122        self.packets_since_ack = self.packets_since_ack.saturating_add(1);
123
124        let mut newly_lost = Vec::new();
125        match self.highest_received {
126            None => self.highest_received = Some(seq_number),
127            Some(highest) if seq::seq_gt(seq_number, highest) => {
128                let mut s = seq::seq_next(highest);
129                let mut n = 0u32;
130                while s != seq_number && n < MAX_GAP_EXPANSION {
131                    newly_lost.push(s);
132                    self.loss_list.insert(s);
133                    s = seq::seq_next(s);
134                    n += 1;
135                }
136                self.highest_received = Some(seq_number);
137            }
138            _ => {}
139        }
140
141        self.loss_list.remove(&seq_number);
142
143        let mut delivered = Vec::new();
144        if seq_number == self.next_expected {
145            delivered.push(seq_number);
146            self.next_expected = seq::seq_next(seq_number);
147            while self.out_of_order.remove(&self.next_expected) {
148                delivered.push(self.next_expected);
149                self.next_expected = seq::seq_next(self.next_expected);
150            }
151        } else if seq::seq_gt(seq_number, self.next_expected) {
152            self.out_of_order.insert(seq_number);
153        }
154        // seq_number before next_expected: a duplicate of an
155        // already-delivered packet (e.g. a redundant retransmission) —
156        // nothing to do.
157
158        let nak = if newly_lost.is_empty() {
159            None
160        } else {
161            Some(self.build_nak(&newly_lost, now))
162        };
163
164        FeedOutcome { delivered, nak }
165    }
166
167    fn build_nak(&self, seqs: &[u32], now: Duration) -> Vec<u8> {
168        let entries = coalesce(seqs);
169        let raw = build_loss_list(&entries).expect("seq numbers are 31-bit by construction");
170        let pkt = ControlPacket::Nak(NakPacket {
171            timestamp: duration_to_wire_us(now),
172            dest_socket_id: self.dest_socket_id,
173            raw_loss_list: &raw,
174        });
175        let mut buf = alloc::vec![0u8; pkt.serialized_len()];
176        pkt.serialize_into(&mut buf)
177            .expect("buffer sized from serialized_len");
178        buf
179    }
180
181    /// Advance to absolute time `now` and emit any periodic control packets
182    /// now due: a Full ACK every [`super::FULL_ACK_PERIOD`] (rule 11), a
183    /// Light ACK once [`super::LIGHT_ACK_THRESHOLD`] packets have arrived
184    /// since the last ACK (rule 12), and a periodic NAK once `NAKInterval`
185    /// has elapsed *and* the loss list is non-empty (rules 21, 22) — never a
186    /// NAK when nothing is believed lost.
187    pub fn tick(&mut self, now: Duration) -> Vec<Vec<u8>> {
188        let mut out = Vec::new();
189
190        if elapsed(now, self.last_full_ack_at) >= FULL_ACK_PERIOD {
191            out.push(self.build_full_ack(now));
192            self.last_full_ack_at = now;
193            self.packets_since_ack = 0;
194        } else if self.packets_since_ack >= LIGHT_ACK_THRESHOLD {
195            out.push(self.build_light_ack());
196            self.packets_since_ack = 0;
197        }
198
199        let interval = nak_interval(self.rtt.rtt(), self.rtt.rtt_var());
200        if !self.loss_list.is_empty() && elapsed(now, self.last_nak_at) >= interval {
201            let seqs: Vec<u32> = self.loss_list.iter().copied().collect();
202            out.push(self.build_nak(&seqs, now));
203            self.last_nak_at = now;
204        }
205
206        out
207    }
208
209    fn build_full_ack(&mut self, now: Duration) -> Vec<u8> {
210        let ack_number = self.next_ack_number;
211        self.next_ack_number = self.next_ack_number.wrapping_add(1);
212        self.outstanding_acks.insert(ack_number, now);
213        let pkt = ControlPacket::Ack(AckPacket {
214            ack_number,
215            timestamp: duration_to_wire_us(now),
216            dest_socket_id: self.dest_socket_id,
217            cif: AckCif::Full {
218                last_ack_seq: self.next_expected,
219                rtt_us: self.rtt.rtt_us(),
220                rtt_var_us: self.rtt.rtt_var_us(),
221                // Bandwidth/rate estimation (§4.7) is out of ARQ scope —
222                // not curated in srt-arq.md, so left at 0 rather than
223                // fabricated.
224                avail_buf_size: 0,
225                pkt_recv_rate: 0,
226                est_link_capacity: 0,
227                recv_rate_bps: 0,
228            },
229        });
230        let mut buf = alloc::vec![0u8; pkt.serialized_len()];
231        pkt.serialize_into(&mut buf)
232            .expect("buffer sized from serialized_len");
233        buf
234    }
235
236    fn build_light_ack(&self) -> Vec<u8> {
237        // §3.2.4: a Light ACK's Acknowledgement Number "should be set to
238        // 0"; it carries no RTT/CIF payload beyond the sequence number
239        // (rule 24).
240        let pkt = ControlPacket::Ack(AckPacket {
241            ack_number: 0,
242            timestamp: 0,
243            dest_socket_id: self.dest_socket_id,
244            cif: AckCif::Light {
245                last_ack_seq: self.next_expected,
246            },
247        });
248        let mut buf = alloc::vec![0u8; pkt.serialized_len()];
249        pkt.serialize_into(&mut buf)
250            .expect("buffer sized from serialized_len");
251        buf
252    }
253
254    /// Process an incoming ACKACK: match it against the outstanding Full ACK
255    /// it acknowledges and update RTT/RTTVar from the round-trip sample
256    /// (rules 26-30). An ACKACK for an unknown/already-matched
257    /// Acknowledgement Number is ignored.
258    pub fn on_ackack(&mut self, ackack: &AckAckPacket, now: Duration) {
259        if let Some(sent_at) = self.outstanding_acks.remove(&ackack.ack_number) {
260            self.rtt.update(elapsed(now, sent_at));
261        }
262    }
263}
264
265/// `now - since`, clamped to zero rather than panicking on a non-monotonic
266/// `now` (a caller bug, not a protocol condition this module needs to
267/// reject).
268fn elapsed(now: Duration, since: Duration) -> Duration {
269    now.checked_sub(since).unwrap_or(Duration::ZERO)
270}
271
272/// Coalesce a run of sequence numbers (already circularly increasing) into
273/// [`LossListEntry`] Single/Range entries (Appendix A) — a compact NAK
274/// encoding; the coalescing itself is not a spec rule.
275fn coalesce(seqs: &[u32]) -> Vec<LossListEntry> {
276    let mut out = Vec::new();
277    let mut i = 0;
278    while i < seqs.len() {
279        let start = seqs[i];
280        let mut end = start;
281        let mut j = i + 1;
282        while j < seqs.len() && seqs[j] == seq::seq_next(end) {
283            end = seqs[j];
284            j += 1;
285        }
286        if start == end {
287            out.push(LossListEntry::Single(start));
288        } else {
289            out.push(LossListEntry::Range(start, end));
290        }
291        i = j;
292    }
293    out
294}
295
296#[cfg(test)]
297mod tests {
298    use super::*;
299
300    const PEER: u32 = 0xBBBB;
301
302    #[test]
303    fn in_order_arrivals_deliver_immediately_without_nak() {
304        let mut r = Receiver::new(PEER, 0);
305        for seq_number in 0..5u32 {
306            let outcome = r.feed_data(seq_number, Duration::ZERO);
307            assert_eq!(outcome.delivered, alloc::vec![seq_number]);
308            assert!(outcome.nak.is_none());
309        }
310        assert_eq!(r.ack_point(), 5);
311        assert_eq!(r.loss_list_len(), 0);
312    }
313
314    #[test]
315    fn a_gap_triggers_an_immediate_nak_and_stalls_delivery() {
316        let mut r = Receiver::new(PEER, 0);
317        r.feed_data(0, Duration::ZERO);
318        r.feed_data(1, Duration::ZERO);
319        let outcome = r.feed_data(3, Duration::ZERO); // seq 2 missing
320        assert!(outcome.delivered.is_empty());
321        let nak = outcome.nak.expect("gap must trigger an immediate NAK");
322        let ControlPacket::Nak(n) = ControlPacket::parse(&nak).unwrap() else {
323            panic!("expected NAK");
324        };
325        let entries: Vec<LossListEntry> = n.entries().map(|e| e.unwrap()).collect();
326        assert_eq!(entries, alloc::vec![LossListEntry::Single(2)]);
327        assert_eq!(r.ack_point(), 2); // stalled at the gap
328        assert_eq!(r.loss_list_len(), 1);
329
330        // Filling the gap unblocks the buffered seq 3 too.
331        let fill = r.feed_data(2, Duration::ZERO);
332        assert_eq!(fill.delivered, alloc::vec![2, 3]);
333        assert!(fill.nak.is_none());
334        assert_eq!(r.ack_point(), 4);
335        assert_eq!(r.loss_list_len(), 0);
336    }
337
338    #[test]
339    fn zero_loss_tick_never_emits_a_nak() {
340        let mut r = Receiver::new(PEER, 0);
341        for seq_number in 0..5u32 {
342            r.feed_data(seq_number, Duration::ZERO);
343        }
344        for ms in 1..200u64 {
345            let out = r.tick(Duration::from_millis(ms));
346            for bytes in &out {
347                assert!(!matches!(
348                    ControlPacket::parse(bytes).unwrap(),
349                    ControlPacket::Nak(_)
350                ));
351            }
352        }
353    }
354
355    #[test]
356    fn full_ack_fires_on_the_10ms_timer_and_light_ack_on_the_64_packet_threshold() {
357        let mut r = Receiver::new(PEER, 0);
358        let out = r.tick(FULL_ACK_PERIOD);
359        assert_eq!(out.len(), 1);
360        let ControlPacket::Ack(ack) = ControlPacket::parse(&out[0]).unwrap() else {
361            panic!("expected ACK");
362        };
363        assert!(matches!(ack.cif, AckCif::Full { .. }));
364        assert_eq!(ack.ack_number, 1);
365
366        // Under the Full ACK period, but past the light-ack packet count.
367        for seq_number in 0..LIGHT_ACK_THRESHOLD {
368            r.feed_data(seq_number, Duration::ZERO);
369        }
370        let out = r.tick(FULL_ACK_PERIOD + Duration::from_millis(1));
371        // Immediately after a Full ACK, the next tick 1ms later is still
372        // under the 10ms period, so a Light ACK fires instead.
373        let out2 = r.tick(FULL_ACK_PERIOD + Duration::from_millis(2));
374        let light =
375            out.into_iter()
376                .chain(out2)
377                .find_map(|b| match ControlPacket::parse(&b).unwrap() {
378                    ControlPacket::Ack(a) if matches!(a.cif, AckCif::Light { .. }) => Some(a),
379                    _ => None,
380                });
381        assert!(
382            light.is_some(),
383            "expected a Light ACK from the 64-packet threshold"
384        );
385    }
386
387    #[test]
388    fn ackack_updates_rtt_from_the_measured_round_trip() {
389        let mut r = Receiver::new(PEER, 0);
390        let out = r.tick(FULL_ACK_PERIOD);
391        let ControlPacket::Ack(ack) = ControlPacket::parse(&out[0]).unwrap() else {
392            panic!("expected ACK");
393        };
394        let ackack = AckAckPacket {
395            ack_number: ack.ack_number,
396            timestamp: 0,
397            dest_socket_id: PEER,
398        };
399        let sample = Duration::from_millis(20);
400        r.on_ackack(&ackack, FULL_ACK_PERIOD + sample);
401        // moved from the 100ms initial value toward the 20ms sample.
402        assert!(r.rtt() < Duration::from_millis(100));
403        assert!(r.rtt() > sample);
404    }
405}
srt_runtime/arq/receiver.rs

srt_runtime/arq/
receiver.rs
