fix(rtp): associate repair SSRC with base stream RTX parameters (closes #12)

rtc/src/peer_connection/handler/endpoint.rs

@@ -11,7 +11,9 @@ use crate::media_stream::track::MediaStreamTrackId;
 use crate::peer_connection::configuration::media_engine::MediaEngine;
 use crate::peer_connection::event::track_event::{RTCTrackEvent, RTCTrackEventInit};
 use crate::rtp_transceiver::rtp_receiver::internal::RTCRtpReceiverInternal;
-use crate::rtp_transceiver::rtp_sender::{RTCRtpCodingParameters, RTCRtpHeaderExtensionCapability};
+use crate::rtp_transceiver::rtp_sender::{
+    RTCRtpCodingParameters, RTCRtpHeaderExtensionCapability, RTCRtpRtxParameters,
+};
 use crate::rtp_transceiver::{RTCRtpReceiverId, SSRC, internal::RTCRtpTransceiverInternal};
 use crate::statistics::accumulator::RTCStatsAccumulator;
 use interceptor::{Interceptor, Packet};
@@ -448,7 +450,60 @@ where
                     .cloned()
             {
                 if !rrid.is_empty() {
-                    //TODO: Add support of handling repair rtp stream id (rrid) #12
+                    // rrid identifies the base stream (rid) that this repair/RTX packet belongs to.
+                    // Associate the repair SSRC with the base stream's RTX parameters.
+                    let has_base_coding =
+                        match receiver.get_coding_parameter_mut_by_rid(rrid.as_str()) {
+                            Some(coding) => {
+                                match coding.rtx.as_mut() {
+                                    Some(rtx) => rtx.ssrc = ssrc,
+                                    None => coding.rtx = Some(RTCRtpRtxParameters { ssrc }),
+                                }
+                                true
+                            }
+                            None => {
+                                warn!(
+                                    "dropping repair/RTX SSRC association: no base coding \
+                                     parameters found for rrid='{}' (repair_ssrc={}, mid='{}', \
+                                     rid='{}')",
+                                    rrid, ssrc, mid, rid,
+                                );
+                                false
+                            }
+                        };
+
+                    if has_base_coding {
+                        // Register the repair stream with the interceptor so RTX
+                        // packets are actually demuxed and forwarded. Use the
+                        // actual packet payload type here: in this branch `codec`
+                        // corresponds to the repair/RTX packet, so looking up an
+                        // RTX PT from `codec.payload_type` would fail (it is
+                        // already the RTX PT).
+                        let parameters = receiver.get_parameters(self.media_engine);
+                        RTCRtpReceiverInternal::interceptor_remote_stream_op(
+                            self.interceptor,
+                            true,
+                            ssrc,
+                            codec.payload_type,
+                            &codec.rtp_codec,
+                            &parameters.rtp_parameters.header_extensions,
+                        );
+
+                        // Update the stats accumulator so RTX packets are
+                        // attributed to the primary stream's stats (the inbound
+                        // stream accumulator may already exist from the base
+                        // stream's OnOpen event).
+                        if let Some(primary_ssrc) = receiver
+                            .get_coding_parameters()
+                            .iter()
+                            .find(|c| c.rid == rrid)
+                            .and_then(|c| c.ssrc)
+                        {
+                            self.stats.update_inbound_rtx_ssrc(primary_ssrc, ssrc);
+                        }
+                    }
+
+                    return Some(receiver.track().track_id().clone());
                 } else {
                     if let Some(coding) = receiver.get_coding_parameter_mut_by_rid(rid.as_str()) {
                         coding.ssrc = Some(ssrc);

rtc/src/statistics/accumulator/mod.rs

@@ -432,6 +432,32 @@ impl RTCStatsAccumulator {
             })
     }
 
+    /// Updates the RTX SSRC association for an existing inbound RTP stream.
+    ///
+    /// This is used when `rrid` (repaired RTP stream ID) arrives after the base
+    /// stream's `InboundRtpStreamAccumulator` has already been created. It updates
+    /// both the reverse-lookup map (`rtx_ssrc_to_primary`) and the inbound stream's
+    /// `rtx_ssrc` field so that `on_rtx_packet_received_if_rtx()` can recognize
+    /// these packets and `getStats()` reports the correct `rtxSsrc`.
+    ///
+    /// # Arguments
+    ///
+    /// * `primary_ssrc` - The SSRC of the base/primary stream
+    /// * `rtx_ssrc` - The RTX SSRC to associate with the primary stream
+    pub(crate) fn update_inbound_rtx_ssrc(&mut self, primary_ssrc: SSRC, rtx_ssrc: SSRC) {
+        if let Some(stream) = self.inbound_rtp_streams.get_mut(&primary_ssrc) {
+            if let Some(old_rtx_ssrc) = stream.rtx_ssrc {
+                if old_rtx_ssrc != rtx_ssrc {
+                    self.rtx_ssrc_to_primary.remove(&old_rtx_ssrc);
+                }
+            }
+
+            stream.rtx_ssrc = Some(rtx_ssrc);
+        }
+
+        self.rtx_ssrc_to_primary.insert(rtx_ssrc, primary_ssrc);
+    }
+
     /// Gets or creates an outbound stream accumulator for the given SSRC.
     ///
     /// # Arguments

rtc/src/statistics/statistics_tests.rs

@@ -1411,3 +1411,90 @@ fn test_stats_selector_transceiver_isolation() {
         );
     }
 }
+
+/// Verifies that `update_inbound_rtx_ssrc` correctly updates both the
+/// `rtx_ssrc_to_primary` reverse-lookup map and the inbound stream
+/// accumulator's `rtx_ssrc` field. This covers the case where `rrid`
+/// arrives after the base stream's accumulator has already been created.
+#[test]
+fn test_update_inbound_rtx_ssrc() {
+    let mut accumulator = RTCStatsAccumulator::new();
+
+    let primary_ssrc: u32 = 1000;
+    let rtx_ssrc: u32 = 2000;
+    let track_id = "track-1";
+    let mid = "0";
+
+    // Create the inbound stream accumulator first (simulating OnOpen for base stream).
+    // Initially, rtx_ssrc is None because the RTX SSRC is not yet known.
+    accumulator.get_or_create_inbound_rtp_streams(
+        primary_ssrc,
+        RtpCodecKind::Video,
+        track_id,
+        mid,
+        None, // rtx_ssrc not known yet
+        None,
+        0,
+    );
+
+    // Verify initial state: rtx_ssrc should be None
+    let stream = accumulator.inbound_rtp_streams.get(&primary_ssrc).unwrap();
+    assert_eq!(stream.rtx_ssrc, None, "rtx_ssrc should initially be None");
+
+    // Now simulate rrid arrival: update the RTX SSRC association
+    accumulator.update_inbound_rtx_ssrc(primary_ssrc, rtx_ssrc);
+
+    // Verify the inbound stream's rtx_ssrc field is updated
+    let stream = accumulator.inbound_rtp_streams.get(&primary_ssrc).unwrap();
+    assert_eq!(
+        stream.rtx_ssrc,
+        Some(rtx_ssrc),
+        "rtx_ssrc should be updated to the repair SSRC"
+    );
+
+    // Verify the reverse lookup map is updated (used by on_rtx_packet_received_if_rtx)
+    // We can test this indirectly: calling on_rtx_packet_received_if_rtx should now
+    // recognize the RTX SSRC and update the primary stream's retransmission stats.
+    accumulator.on_rtx_packet_received_if_rtx(rtx_ssrc, 50);
+
+    let stream = accumulator.inbound_rtp_streams.get(&primary_ssrc).unwrap();
+    assert_eq!(
+        stream.retransmitted_packets_received, 1,
+        "RTX packet should be attributed to primary stream via reverse lookup"
+    );
+    assert_eq!(
+        stream.retransmitted_bytes_received, 50,
+        "RTX bytes should be attributed to primary stream"
+    );
+}
+
+/// Verifies that `update_inbound_rtx_ssrc` is a no-op when the primary
+/// SSRC does not have an existing inbound stream accumulator (the reverse
+/// lookup is still populated for future use).
+#[test]
+fn test_update_inbound_rtx_ssrc_no_existing_stream() {
+    let mut accumulator = RTCStatsAccumulator::new();
+
+    let primary_ssrc: u32 = 3000;
+    let rtx_ssrc: u32 = 4000;
+
+    // Call update without creating the inbound stream first
+    accumulator.update_inbound_rtx_ssrc(primary_ssrc, rtx_ssrc);
+
+    // The inbound stream should NOT exist
+    assert!(
+        !accumulator.inbound_rtp_streams.contains_key(&primary_ssrc),
+        "No inbound stream should be created by update_inbound_rtx_ssrc"
+    );
+
+    // But the reverse lookup should still be populated so that when the
+    // inbound stream is created later, RTX packets can be attributed.
+    assert_eq!(
+        accumulator.rtx_ssrc_to_primary.get(&rtx_ssrc),
+        Some(&primary_ssrc),
+        "Reverse lookup should map RTX SSRC to the primary SSRC"
+    );
+
+    // Verify the reverse lookup can be used without panicking.
+    accumulator.on_rtx_packet_received_if_rtx(rtx_ssrc, 50);
+}