feat/remote track republish

CHANGELOG.md

@@ -78,7 +78,9 @@ publish new tracks.
 
 `write_pcm` queues up to `LocalAudioTrackConfig::pcm_queue_capacity`, 60 s by
 default (it was 200 ms). A producer that writes faster than real time gets
-`PcmQueueOverflow` only when the queue is full.
+`PcmQueueOverflow` only when the queue is full. The capacity must hold at least
+one 20 ms frame; `LocalAudioTrack::opus_with_config` returns `RtcError::Media`
+for a smaller value.
 
 ## New Features
 
@@ -102,6 +104,14 @@ media of the new track. Before, a second video publish failed with
 cannot replace or be replaced: `publish` returns `RtcError::SimulcastReplace`.
 Turn such a track off and on with `mute_track` and `unmute_track`.
 
+### A dropped remote track comes back after a republish
+
+Dropping a `RemoteTrack` unsubscribes it. When the publisher publishes the
+track again (for example after an unmute), `on_track` now delivers a new
+`RemoteTrack` for it, also when the SFU sends the media on the receiver that
+the dropped track used. Before, the track did not come back. A track that is
+kept through a mute gets the media again and is not delivered a second time.
+
 ### A token-only client can prepare a call before the join
 
 `RtcClient::call` returns an `RtcCall` that is not joined yet, and
@@ -136,6 +146,34 @@ on `Call` (`get_call_participant_session_metrics`,
 `query_call_session_participant_stats`,
 `get_call_session_participant_stats_timeline`).
 
+## Fixes
+
+- After a join or a reconnect, `sfu_events()` gives `ParticipantJoined`,
+  `ParticipantUpdated` and `ParticipantLeft` only for the participants that
+  changed since the previous join response. Before, each join response sent
+  `ParticipantJoined` for every participant. A change of speaking state, audio
+  level, connection quality or dominant speaker does not count: its own event
+  reports it.
+- Each join starts with an empty participant list and call state, also after
+  a join that failed or was dropped. A new join no longer reports
+  `ParticipantLeft` for the participants of the earlier attempt.
+- `ParticipantCountChanged` comes only when the count changes, not with each
+  SFU health check.
+- A dropped `RemoteTrack` unsubscribes its track only when it is the latest
+  track for that participant and track type. It can be dropped on a thread
+  without a Tokio runtime.
+- A dropped `join` future sets the call back to `Idle`, so the next `join` is
+  accepted. A dropped `leave` future still sends the leave, closes the
+  connection and sets `Left`.
+- The first join retry waits 250–500 ms and a later one up to 2.5 s, as in
+  stream-video-js.
+- The pacer task of a `LocalAudioTrack` ends when the track is dropped. A
+  decoded `PcmFrame` holds only its own samples.
+- A muted audio track no longer encodes its PCM.
+- While a track is muted, its RTP clock keeps running, for audio and video. The
+  first packet after an unmute shows the length of the mute (RFC 3550). Before,
+  the timestamps continued from the last packet before the mute.
+
 # v0.1.0-preview.2
 
 docs.rs builds on current nightly. `doc_auto_cfg` was removed in 1.92 and

src/rtc/join/connection.rs

@@ -140,8 +140,8 @@ pub(super) fn register_on_track(
                 reconnect_enabled,
                 track,
                 weak_pc,
-            )
-            .await;
+                None,
+            );
         })
     }));
 }

src/rtc/join/mod.rs

@@ -71,7 +71,7 @@ use super::subscriptions::{
     DEFAULT_VIDEO_DIMENSION, SubscriptionConfig, SubscriptionTarget, TrackKey,
 };
 use super::tracer::Tracer;
-use super::tracks::{LocalTrack, RemoteParticipant, RemoteTrack};
+use super::tracks::{LocalTrack, RemoteParticipant, RemoteTrack, RtpPacket};
 
 use serde_json::json;
 

src/rtc/join/subscriptions_runtime.rs

@@ -182,14 +182,15 @@ impl RtcCore {
     /// Correlate an inbound track to a participant, build a [`RemoteTrack`], and
     /// deliver it to the `on_track` callback. Called by the subscriber PC, which
     /// passes itself as `subscriber` so the track can send RTCP keyframe
-    /// requests.
-    pub(super) async fn handle_incoming_track(
+    /// requests. `first_packet` is a packet already read from `track`.
+    pub(super) fn handle_incoming_track(
         self: Arc<Self>,
         generation: u64,
         connection_epoch: u64,
         reconnect_enabled: Arc<AtomicBool>,
         track: Arc<TrackRemote>,
         subscriber: Weak<RTCPeerConnection>,
+        first_packet: Option<RtpPacket>,
     ) {
         if !self.is_generation_current(generation) || !reconnect_enabled.load(Ordering::SeqCst) {
             return;
@@ -231,44 +232,71 @@ impl RtcCore {
             .unwrap_or_else(|e| e.into_inner())
             .insert(key.clone(), track_id);
         let weak = Arc::downgrade(&self);
+        let (kept_track, kept_subscriber) = (track.clone(), subscriber.clone());
         // The caller can drop the track on a thread without a runtime.
         let runtime = tokio::runtime::Handle::current();
         let on_drop = Box::new(move || {
             if let Some(core) = weak.upgrade() {
                 let _runtime = runtime.enter();
                 let task_core = core.clone();
                 std::mem::drop(core.spawn_generation_task(generation, async move {
-                    task_core
-                        .on_remote_track_dropped(generation, connection_epoch, key, track_id)
-                        .await;
+                    if task_core
+                        .clone()
+                        .on_remote_track_dropped(
+                            generation,
+                            connection_epoch,
+                            key.clone(),
+                            track_id,
+                        )
+                        .await
+                    {
+                        task_core
+                            .deliver_again_when_published(
+                                generation,
+                                connection_epoch,
+                                reconnect_enabled,
+                                key,
+                                kept_track,
+                                kept_subscriber,
+                            )
+                            .await;
+                    }
                 }));
             }
         });
-        let remote = RemoteTrack::new(track, participant, track_type, subscriber, on_drop);
+        let remote = RemoteTrack::new(
+            track,
+            participant,
+            track_type,
+            subscriber,
+            on_drop,
+            first_packet,
+        );
         cb(remote);
     }
 
-    /// The publisher dropped their inbound track handle → unsubscribe from it.
+    /// The consumer dropped their inbound track handle → unsubscribe from it.
+    /// Returns whether the dropped track was the latest one for its key.
     pub(super) async fn on_remote_track_dropped(
         self: Arc<Self>,
         generation: u64,
         connection_epoch: u64,
         key: TrackKey,
         track_id: u64,
-    ) {
+    ) -> bool {
         {
             let connection = self.connection.lock().await;
             if !connection.as_ref().is_some_and(|current| {
                 current.generation == generation && current.epoch == connection_epoch
             }) {
-                return;
+                return false;
             }
             let mut delivered = self
                 .delivered_tracks
                 .lock()
                 .unwrap_or_else(|e| e.into_inner());
             if delivered.get(&key) != Some(&track_id) {
-                return;
+                return false;
             }
             delivered.remove(&key);
             drop(delivered);
@@ -281,10 +309,59 @@ impl RtcCore {
             .is_connection_current(generation, connection_epoch)
             .await
         {
-            return;
+            return false;
         }
         if let Err(e) = self.recompute_subscriptions().await {
             tracing::debug!(error = %e, "stream.rtc.unsubscribe_on_drop_failed");
         }
+        true
+    }
+
+    /// Read and drop the packets of a dropped track. When a packet arrives
+    /// after a republish, deliver the track again with that packet first:
+    /// webrtc-rs fires `on_track` only once for each receiver. The SFU can
+    /// send the first packets before its answer to the new subscription.
+    async fn deliver_again_when_published(
+        self: Arc<Self>,
+        generation: u64,
+        connection_epoch: u64,
+        reconnect_enabled: Arc<AtomicBool>,
+        key: TrackKey,
+        track: Arc<TrackRemote>,
+        subscriber: Weak<RTCPeerConnection>,
+    ) {
+        while let Ok((packet, _)) = track.read_rtp().await {
+            // A track that a new receiver delivered, or a new connection, ends
+            // the wait.
+            let replaced = self
+                .delivered_tracks
+                .lock()
+                .unwrap_or_else(|e| e.into_inner())
+                .contains_key(&key);
+            if replaced
+                || !self
+                    .is_connection_current(generation, connection_epoch)
+                    .await
+            {
+                return;
+            }
+            // The drop set `manual_unsub`, and only a republish clears it.
+            let dropped = self
+                .manual_unsub
+                .lock()
+                .unwrap_or_else(|e| e.into_inner())
+                .contains(&key);
+            if !dropped {
+                self.handle_incoming_track(
+                    generation,
+                    connection_epoch,
+                    reconnect_enabled,
+                    track,
+                    subscriber,
+                    Some(packet),
+                );
+                return;
+            }
+        }
     }
 }

src/rtc/join/tests.rs

@@ -1177,15 +1177,14 @@ async fn only_the_latest_remote_track_unsubscribes_when_dropped_without_a_runtim
         .expect("inbound track")
         .expect("inbound track channel");
     for _ in 0..2 {
-        core.clone()
-            .handle_incoming_track(
-                generation,
-                epoch,
-                reconnect_enabled.clone(),
-                inbound.clone(),
-                Arc::downgrade(&receiver),
-            )
-            .await;
+        core.clone().handle_incoming_track(
+            generation,
+            epoch,
+            reconnect_enabled.clone(),
+            inbound.clone(),
+            Arc::downgrade(&receiver),
+            None,
+        );
     }
     let stale = remote_rx.recv().expect("stale remote track");
     let remote = remote_rx.recv().expect("latest remote track");
@@ -1231,6 +1230,68 @@ async fn only_the_latest_remote_track_unsubscribes_when_dropped_without_a_runtim
     core.leave("test cleanup").await.expect("leave");
 }
 
+#[tokio::test]
+async fn a_dropped_track_is_delivered_again_only_after_it_is_published_again() {
+    let core = test_core();
+    let generation = prepare_joined_core(&core, "alice");
+    let (connection, _sfu) = establish_fake(&core, generation).await;
+    let (epoch, reconnect_enabled) = (connection.epoch, connection.reconnect_enabled.clone());
+    *core.connection.lock().await = Some(connection);
+    let (remote_tx, mut remote_rx) = tokio::sync::mpsc::unbounded_channel();
+    *core
+        .on_track_cb
+        .lock()
+        .unwrap_or_else(|error| error.into_inner()) = Some(Arc::new(move |remote| {
+        let _ = remote_tx.send(remote);
+    }));
+    let audio = LocalAudioTrack::opus().expect("opus track");
+    let sender = peer::new_peer_connection(&[]).await.expect("sender");
+    sender
+        .add_track(audio.webrtc_track())
+        .await
+        .expect("add track");
+    let (receiver, mut inbound) = peer::connect_audio_receiver(&sender).await;
+    audio.start_pacing().await;
+    let inbound = tokio::time::timeout(Duration::from_secs(5), inbound.recv())
+        .await
+        .expect("inbound track")
+        .expect("inbound track channel");
+    core.clone().handle_incoming_track(
+        generation,
+        epoch,
+        reconnect_enabled,
+        inbound,
+        Arc::downgrade(&receiver),
+        None,
+    );
+    let remote = remote_rx.recv().await.expect("remote track");
+    let session_id = remote.participant().session_id.clone();
+
+    // The publisher keeps sending after the drop.
+    drop(remote);
+    assert!(
+        tokio::time::timeout(Duration::from_millis(500), remote_rx.recv())
+            .await
+            .is_err(),
+        "a dropped live track does not come back"
+    );
+
+    core.add_published_track("bob", &session_id, TrackType::Audio as i32, None);
+    let again = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv())
+        .await
+        .expect("the track is delivered again after the republish")
+        .expect("remote track");
+    tokio::time::timeout(Duration::from_secs(5), again.next_pcm())
+        .await
+        .expect("audio of the track delivered again")
+        .expect("pcm frame");
+
+    audio.stop();
+    let _ = sender.close().await;
+    let _ = receiver.close().await;
+    core.leave("test cleanup").await.expect("leave");
+}
+
 #[tokio::test]
 async fn detached_connection_ignores_publish_options_from_its_sfu() {
     let core = test_core();

src/rtc/tracks/remote.rs

@@ -238,7 +238,8 @@ enum Decode {
 /// An inbound media track from a remote participant.
 ///
 /// Not `Clone`: it owns the inbound stream and unsubscribes on drop. Wrap it in
-/// an `Arc` if you need shared read handles.
+/// an `Arc` if you need shared read handles. When the publisher publishes a
+/// dropped track again, `on_track` delivers a new `RemoteTrack` for it.
 pub struct RemoteTrack {
     track: Arc<TrackRemote>,
     participant: RemoteParticipant,
@@ -258,20 +259,24 @@ pub struct RemoteTrack {
     last_keyframe_request: StdMutex<Option<Instant>>,
     /// Invoked once on drop to unsubscribe from the SFU.
     on_drop: StdMutex<Option<Box<dyn FnOnce() + Send>>>,
+    /// A packet read before this track was built, returned by the first read.
+    first_packet: StdMutex<Option<RtpPacket>>,
 }
 
 impl RemoteTrack {
     /// Build a `RemoteTrack` from a subscriber `on_track` event.
     ///
     /// `subscriber` is the PeerConnection the track arrived on, used to send
     /// RTCP keyframe requests. `on_drop` is invoked exactly once when the track
-    /// is dropped so the call can retract the subscription.
+    /// is dropped so the call can retract the subscription. `first_packet` is
+    /// returned before the packets of `track`.
     pub(crate) fn new(
         track: Arc<TrackRemote>,
         participant: RemoteParticipant,
         track_type: TrackType,
         subscriber: Weak<RTCPeerConnection>,
         on_drop: Box<dyn FnOnce() + Send>,
+        first_packet: Option<RtpPacket>,
     ) -> Self {
         let params = track.codec();
         let codec = Codec {
@@ -294,6 +299,7 @@ impl RemoteTrack {
             subscriber,
             last_keyframe_request: StdMutex::new(None),
             on_drop: StdMutex::new(Some(on_drop)),
+            first_packet: StdMutex::new(first_packet),
         }
     }
 
@@ -317,6 +323,7 @@ impl RemoteTrack {
             track_type,
             Arc::downgrade(peer),
             Box::new(|| {}),
+            None,
         )
     }
 
@@ -353,6 +360,14 @@ impl RemoteTrack {
     }
 
     async fn read_rtp_inner(&self) -> Option<RtpPacket> {
+        let first = self
+            .first_packet
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .take();
+        if first.is_some() {
+            return first;
+        }
         match self.track.read_rtp().await {
             Ok((pkt, _attr)) => Some(pkt),
             Err(e) => {