rtc_interceptor/twcc/
receiver.rs

1//! TWCC Receiver Interceptor - tracks incoming packets and generates feedback.
2
3use super::recorder::Recorder;
4use super::stream_supports_twcc;
5use crate::Interceptor;
6use crate::stream_info::StreamInfo;
7use crate::{AttributedPacket, Packet, TaggedPacket};
8use sansio::Protocol;
9use shared::TransportContext;
10use shared::error::Error;
11use shared::marshal::Unmarshal;
12use std::collections::{HashMap, VecDeque};
13use std::time::{Duration, Instant};
14
15/// Default interval for sending TWCC feedback.
16const DEFAULT_INTERVAL: Duration = Duration::from_millis(100);
17
18/// Builder for the TwccReceiverInterceptor.
19///
20/// # Example
21///
22/// ```
23/// use rtc_interceptor::{Slot, Registry, TwccReceiverBuilder};
24/// use std::time::Duration;
25///
26/// let chain = Registry::new()
27///     .with(Slot::TwccReceiver, TwccReceiverBuilder::new()
28///         .with_interval(Duration::from_millis(100))
29///         .build())
30///     .build();
31/// ```
32pub struct TwccReceiverBuilder {
33    /// Interval between feedback reports.
34    interval: Duration,
35}
36
37impl Default for TwccReceiverBuilder {
38    fn default() -> Self {
39        Self {
40            interval: DEFAULT_INTERVAL,
41        }
42    }
43}
44
45impl TwccReceiverBuilder {
46    /// Create a new builder with default settings.
47    pub fn new() -> Self {
48        Self::default()
49    }
50
51    /// Set the interval between feedback reports.
52    pub fn with_interval(mut self, interval: Duration) -> Self {
53        self.interval = interval;
54        self
55    }
56
57    /// Build the interceptor.
58    pub fn build(self) -> TwccReceiverInterceptor {
59        TwccReceiverInterceptor::new(self.interval)
60    }
61}
62
63/// Per-stream state for the receiver.
64struct RemoteStream {
65    /// Header extension ID for transport-wide CC.
66    hdr_ext_id: u8,
67}
68
69/// Interceptor that tracks incoming RTP packets and generates TWCC feedback.
70///
71/// This interceptor examines incoming RTP packets for transport-wide CC sequence
72/// numbers and periodically generates TransportLayerCC feedback packets.
73pub struct TwccReceiverInterceptor {
74    /// Configuration
75    interval: Duration,
76
77    /// Start time for calculating arrival times.
78    start_time: Option<Instant>,
79
80    /// TWCC recorder for building feedback.
81    recorder: Option<Recorder>,
82
83    /// Remote stream state per SSRC.
84    streams: HashMap<u32, RemoteStream>,
85
86    /// Transport-wide CC header extension ID negotiated on this transport.
87    ///
88    /// Transport-wide congestion control is per *transport*, not per SSRC: the feedback
89    /// must account for every packet that carried the transport-cc sequence extension,
90    /// whatever SSRC it rode on. Packets can legitimately arrive on SSRCs that have no
91    /// stream binding - RTX/probe padding and RID simulcast layers before they are bound -
92    /// and omitting them makes the remote congestion controller read them as lost. This ID
93    /// is used as the fallback for those unbound SSRCs.
94    transport_hdr_ext_id: Option<u8>,
95
96    /// Queue for feedback packets.
97    write_queue: VecDeque<TaggedPacket>,
98
99    /// Next timeout for sending feedback.
100    next_timeout: Option<Instant>,
101    /// Inbound packets ready for the next interceptor.
102    read_queue: VecDeque<TaggedPacket>,
103}
104
105impl TwccReceiverInterceptor {
106    fn new(interval: Duration) -> Self {
107        Self {
108            read_queue: VecDeque::new(),
109            interval,
110            start_time: None,
111            recorder: None,
112            streams: HashMap::new(),
113            transport_hdr_ext_id: None,
114            write_queue: VecDeque::new(),
115            next_timeout: None,
116        }
117    }
118
119    fn generate_feedback(&mut self, now: Instant) {
120        let Some(recorder) = self.recorder.as_mut() else {
121            return;
122        };
123
124        let packets = recorder.build_feedback_packet();
125        for pkt in packets {
126            self.write_queue.push_back(TaggedPacket {
127                now,
128                transport: TransportContext::default(),
129                message: AttributedPacket::new(Packet::Rtcp(vec![pkt])),
130            });
131        }
132    }
133}
134
135impl Protocol<TaggedPacket, TaggedPacket, ()> for TwccReceiverInterceptor {
136    type Rout = TaggedPacket;
137    type Wout = TaggedPacket;
138    type Eout = ();
139    type Error = Error;
140    type Time = Instant;
141
142    fn handle_read(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
143        // Process incoming RTP packets with TWCC extension
144        if let Packet::Rtp(ref rtp_packet) = msg.message.packet {
145            // Prefer the ID bound for this SSRC, and fall back to the transport-wide one so
146            // that padding and not-yet-bound streams still make it into the feedback.
147            let hdr_ext_id = self
148                .streams
149                .get(&rtp_packet.header.ssrc)
150                .map(|stream| stream.hdr_ext_id)
151                .or(self.transport_hdr_ext_id);
152
153            // Extract transport CC sequence number
154            if let Some(hdr_ext_id) = hdr_ext_id
155                && let Some(ext_data) = rtp_packet.header.get_extension(hdr_ext_id)
156                && let Ok(tcc) =
157                    rtp::extension::transport_cc_extension::TransportCcExtension::unmarshal(
158                        &mut ext_data.as_ref(),
159                    )
160            {
161                // Initialize recorder on the first packet carrying transport-wide CC
162                if self.recorder.is_none() {
163                    // Use a random sender SSRC for feedback
164                    self.recorder = Some(Recorder::new(rand::random()));
165                    self.start_time = Some(msg.now);
166                    self.next_timeout = Some(msg.now + self.interval);
167                }
168
169                // Calculate arrival time in microseconds since start
170                let arrival_time = self
171                    .start_time
172                    .map(|start| msg.now.duration_since(start).as_micros() as i64)
173                    .unwrap_or(0);
174
175                if let Some(recorder) = self.recorder.as_mut() {
176                    recorder.record(rtp_packet.header.ssrc, tcc.transport_sequence, arrival_time);
177                }
178            }
179        }
180
181        self.read_queue.push_back(msg);
182
183        Ok(())
184    }
185
186    fn poll_read(&mut self) -> Option<Self::Rout> {
187        self.read_queue.pop_front()
188    }
189
190    fn handle_write(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
191        self.write_queue.push_back(msg);
192        Ok(())
193    }
194
195    fn poll_write(&mut self) -> Option<TaggedPacket> {
196        // First drain feedback packets
197        if let Some(pkt) = self.write_queue.pop_front() {
198            return Some(pkt);
199        }
200        None
201    }
202
203    fn handle_timeout(&mut self, now: Instant) -> Result<(), Error> {
204        // Check if we need to send feedback
205        if let Some(timeout) = self.next_timeout
206            && now >= timeout
207        {
208            self.generate_feedback(now);
209            self.next_timeout = Some(now + self.interval);
210        }
211        Ok(())
212    }
213
214    fn poll_timeout(&mut self) -> Option<Instant> {
215        self.next_timeout
216    }
217}
218
219impl Interceptor for TwccReceiverInterceptor {
220    fn bind_remote_stream(&mut self, info: &StreamInfo) {
221        if let Some(hdr_ext_id) = stream_supports_twcc(info) {
222            // Don't track if ID is 0 (invalid)
223            if hdr_ext_id != 0 {
224                self.streams.insert(info.ssrc, RemoteStream { hdr_ext_id });
225                // An extension ID maps to exactly one URI within an RTP session, and BUNDLE
226                // makes the bundled m-lines one session, so every binding here carries the
227                // same transport-cc ID: taking the newest is taking the only one. A genuinely
228                // different ID means renegotiation, where the newest is also what we want.
229                self.transport_hdr_ext_id = Some(hdr_ext_id);
230            }
231        }
232    }
233
234    fn unbind_remote_stream(&mut self, info: &StreamInfo) {
235        self.streams.remove(&info.ssrc);
236        // Any binding that remains carries the same ID (see `bind_remote_stream`), so only the
237        // loss of the last one leaves no negotiated ID to fall back on. Dropping it then keeps
238        // a stale ID from outliving the negotiation and being misread as transport-cc.
239        if self.streams.is_empty() {
240            self.transport_hdr_ext_id = None;
241        }
242    }
243
244    fn bind_local_stream(&mut self, _info: &StreamInfo) {}
245
246    fn unbind_local_stream(&mut self, _info: &StreamInfo) {}
247}
rtc_interceptor/twcc/receiver.rs

rtc_interceptor/twcc/
receiver.rs
