Merge pull request #316 from webrtc-rs/fix/track-binding-pause

Pause in TrackBinding when direction is not send

Author: algesten

webrtc/CHANGELOG.md

@@ -5,7 +5,8 @@
 * Added more stats to `RemoteInboundRTPStats` and `RemoteOutboundRTPStats` [#282](https://github.com/webrtc-rs/webrtc/pull/282) by [@k0nserv](https://github.com/k0nserv).
 * Don't register `video/rtx` codecs in `MediaEngine::register_default_codecs`. These weren't actually support and prevented RTX in the existing RTP stream from being used. Long term we should support RTX via this method, this is tracked in [#295](https://github.com/webrtc-rs/webrtc/issues/295). [#294 Remove video/rtx codecs](https://github.com/webrtc-rs/webrtc/pull/294) contributed by [k0nserv](https://github.com/k0nserv)
 * Add IP filter to WebRTC `SettingEngine` [#306](https://github.com/webrtc-rs/webrtc/pull/306)
-
+* Stop sequence numbers from increasing in `TrackLocalStaticSample` while the bound `RTCRtpSender` have
+directions that should not send. [#316](https://github.com/webrtc-rs/webrtc/pull/316)
 
 ## 0.5.1
 

webrtc/src/rtp_transceiver/rtp_sender/mod.rs

@@ -332,6 +332,7 @@ impl RTCRtpSender {
                     .await,
                 ssrc: context.ssrc,
                 write_stream: context.write_stream.clone(),
+                paused: self.paused.clone(),
             };
 
             t.bind(&new_context).await
@@ -392,6 +393,7 @@ impl RTCRtpSender {
                 write_stream: Some(
                     Arc::clone(&write_stream) as Arc<dyn TrackLocalWriter + Send + Sync>
                 ),
+                paused: self.paused.clone(),
             };
 
             let codec = if let Some(t) = &*track {

webrtc/src/track/track_local/mod.rs

@@ -35,6 +35,7 @@ pub struct TrackLocalContext {
     pub(crate) params: RTCRtpParameters,
     pub(crate) ssrc: SSRC,
     pub(crate) write_stream: Option<Arc<dyn TrackLocalWriter + Send + Sync>>,
+    pub(crate) paused: Arc<AtomicBool>,
 }
 
 impl TrackLocalContext {
@@ -104,20 +105,31 @@ pub(crate) struct TrackBinding {
     ssrc: SSRC,
     payload_type: PayloadType,
     write_stream: Option<Arc<dyn TrackLocalWriter + Send + Sync>>,
+    sender_paused: Arc<AtomicBool>,
+}
+
+impl TrackBinding {
+    pub fn is_sender_paused(&self) -> bool {
+        self.sender_paused.load(Ordering::SeqCst)
+    }
 }
 
 pub(crate) struct InterceptorToTrackLocalWriter {
     pub(crate) interceptor_rtp_writer: Mutex<Option<Arc<dyn RTPWriter + Send + Sync>>>,
-    paused: Arc<AtomicBool>,
+    sender_paused: Arc<AtomicBool>,
 }
 
 impl InterceptorToTrackLocalWriter {
     pub(crate) fn new(paused: Arc<AtomicBool>) -> Self {
         InterceptorToTrackLocalWriter {
             interceptor_rtp_writer: Mutex::new(None),
-            paused,
+            sender_paused: paused,
         }
     }
+
+    fn is_sender_paused(&self) -> bool {
+        self.sender_paused.load(Ordering::SeqCst)
+    }
 }
 
 impl std::fmt::Debug for InterceptorToTrackLocalWriter {
@@ -129,7 +141,7 @@ impl std::fmt::Debug for InterceptorToTrackLocalWriter {
 #[async_trait]
 impl TrackLocalWriter for InterceptorToTrackLocalWriter {
     async fn write_rtp(&self, pkt: &rtp::packet::Packet) -> Result<usize> {
-        if self.paused.load(Ordering::SeqCst) {
+        if self.is_sender_paused() {
             return Ok(0);
         }
 

webrtc/src/track/track_local/track_local_static_rtp.rs

@@ -28,6 +28,20 @@ impl TrackLocalStaticRTP {
     pub fn codec(&self) -> RTCRtpCodecCapability {
         self.codec.clone()
     }
+
+    pub async fn any_binding_paused(&self) -> bool {
+        let bindings = self.bindings.lock().await;
+        bindings
+            .iter()
+            .any(|b| b.sender_paused.load(Ordering::SeqCst))
+    }
+
+    pub async fn all_binding_paused(&self) -> bool {
+        let bindings = self.bindings.lock().await;
+        bindings
+            .iter()
+            .all(|b| b.sender_paused.load(Ordering::SeqCst))
+    }
 }
 
 #[async_trait]
@@ -50,6 +64,7 @@ impl TrackLocal for TrackLocalStaticRTP {
                     payload_type: codec.payload_type,
                     write_stream: t.write_stream(),
                     id: t.id(),
+                    sender_paused: t.paused.clone(),
                 }));
             }
 
@@ -112,6 +127,11 @@ impl TrackLocalWriter for TrackLocalStaticRTP {
     /// If one PeerConnection fails the packets will still be sent to
     /// all PeerConnections. The error message will contain the ID of the failed
     /// PeerConnections so you can remove them
+    ///
+    /// If the RTCRtpSender direction is such that no packets should be sent, any call to this
+    /// function are blocked internally. Care must be taken to not increase the sequence number
+    /// while the sender is paused. While the actual _sending_ is blocked, the receiver will
+    /// miss out when the sequence number "rolls over", which in turn will break SRTP.
     async fn write_rtp(&self, p: &rtp::packet::Packet) -> Result<usize> {
         let mut n = 0;
         let mut write_errs = vec![];
@@ -122,6 +142,10 @@ impl TrackLocalWriter for TrackLocalStaticRTP {
             bindings.clone()
         };
         for b in bindings {
+            if b.is_sender_paused() {
+                // See caveat in function doc.
+                continue;
+            }
             pkt.header.ssrc = b.ssrc;
             pkt.header.payload_type = b.payload_type;
             if let Some(write_stream) = &b.write_stream {

webrtc/src/track/track_local/track_local_static_sample.rs

@@ -3,6 +3,7 @@ use super::*;
 use crate::error::flatten_errs;
 
 use crate::track::RTP_OUTBOUND_MTU;
+use log::warn;
 use media::Sample;
 use tokio::sync::Mutex;
 
@@ -11,6 +12,7 @@ struct TrackLocalStaticSampleInternal {
     packetizer: Option<Box<dyn rtp::packetizer::Packetizer + Send + Sync>>,
     sequencer: Option<Box<dyn rtp::sequence::Sequencer + Send + Sync>>,
     clock_rate: f64,
+    did_warn_about_wonky_pause: bool,
 }
 
 /// TrackLocalStaticSample is a TrackLocal that has a pre-set codec and accepts Samples.
@@ -32,6 +34,7 @@ impl TrackLocalStaticSample {
                 packetizer: None,
                 sequencer: None,
                 clock_rate: 0.0f64,
+                did_warn_about_wonky_pause: false,
             }),
         }
     }
@@ -52,6 +55,42 @@ impl TrackLocalStaticSample {
             return Ok(());
         }
 
+        let (any_paused, all_paused) = (
+            self.rtp_track.any_binding_paused().await,
+            self.rtp_track.all_binding_paused().await,
+        );
+
+        if all_paused {
+            // Abort already here to not increment sequence numbers.
+            return Ok(());
+        }
+
+        if any_paused {
+            // This is a problem state due to how this impl is structured. The sequencer will allocate
+            // one sequence number per RTP packet regardless of how many TrackBinding that will send
+            // the packet. I.e. we get the same sequence number per multiple SSRC, which is not good
+            // for SRTP, but that's how it works.
+            //
+            // SRTP has a further problem with regards to jumps in sequence number. Consider this:
+            //
+            // 1. Create track local
+            // 2. Bind track local to track 1.
+            // 3. Bind track local to track 2.
+            // 4. Pause track 1.
+            // 5. Keep sending...
+            //
+            // At this point, the track local will keep incrementing the sequence number, because we have
+            // one binding that is still active. However SRTP hmac verifying (tag), can only accept a
+            // relatively small jump in sequence numbers since it uses the ROC (i.e. how many times the
+            // sequence number has rolled over), which means if this pause state of one binding persists
+            // for a longer time, the track can never be resumed since the receiver would have missed
+            // the rollovers.
+            if !internal.did_warn_about_wonky_pause {
+                internal.did_warn_about_wonky_pause = true;
+                warn!("Detected multiple track bindings where only one was paused");
+            }
+        }
+
         // skip packets by the number of previously dropped packets
         if let Some(sequencer) = &internal.sequencer {
             for _ in 0..sample.prev_dropped_packets {