diff --git a/CHANGELOG.md b/CHANGELOG.md index 91e3c39..893827d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/rtc/join/connection.rs b/src/rtc/join/connection.rs index 5733af4..942d853 100644 --- a/src/rtc/join/connection.rs +++ b/src/rtc/join/connection.rs @@ -140,8 +140,8 @@ pub(super) fn register_on_track( reconnect_enabled, track, weak_pc, - ) - .await; + None, + ); }) })); } diff --git a/src/rtc/join/mod.rs b/src/rtc/join/mod.rs index 800d206..4b3b805 100644 --- a/src/rtc/join/mod.rs +++ b/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; diff --git a/src/rtc/join/subscriptions_runtime.rs b/src/rtc/join/subscriptions_runtime.rs index a564828..8c3bbb4 100644 --- a/src/rtc/join/subscriptions_runtime.rs +++ b/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, generation: u64, connection_epoch: u64, reconnect_enabled: Arc, track: Arc, subscriber: Weak, + first_packet: Option, ) { if !self.is_generation_current(generation) || !reconnect_enabled.load(Ordering::SeqCst) { return; @@ -231,6 +232,7 @@ 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 || { @@ -238,37 +240,63 @@ impl RtcCore { 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, 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, + generation: u64, + connection_epoch: u64, + reconnect_enabled: Arc, + key: TrackKey, + track: Arc, + subscriber: Weak, + ) { + 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; + } + } } } diff --git a/src/rtc/join/tests.rs b/src/rtc/join/tests.rs index 6810d35..c19817b 100644 --- a/src/rtc/join/tests.rs +++ b/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(); diff --git a/src/rtc/tracks/remote.rs b/src/rtc/tracks/remote.rs index 77fc89d..72f13c1 100644 --- a/src/rtc/tracks/remote.rs +++ b/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, participant: RemoteParticipant, @@ -258,6 +259,8 @@ pub struct RemoteTrack { last_keyframe_request: StdMutex>, /// Invoked once on drop to unsubscribe from the SFU. on_drop: StdMutex>>, + /// A packet read before this track was built, returned by the first read. + first_packet: StdMutex>, } impl RemoteTrack { @@ -265,13 +268,15 @@ impl RemoteTrack { /// /// `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, participant: RemoteParticipant, track_type: TrackType, subscriber: Weak, on_drop: Box, + first_packet: Option, ) -> 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 { + 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) => { diff --git a/tests/rtc_media.rs b/tests/rtc_media.rs index e4f35f1..e85930b 100644 --- a/tests/rtc_media.rs +++ b/tests/rtc_media.rs @@ -751,7 +751,7 @@ async fn no_video_on_track_without_video_subscription() { .await .expect("A publish video"); - // B subscribes to audio only (the default). + // B subscribes to audio only. let mut rx_b = track_sink(&call_b); call_b .join(JoinCallData::new(&user_b)) @@ -1609,3 +1609,374 @@ async fn sole_video_can_be_stopped_and_published_again() { let _ = admin.delete(DeleteCallRequest { hard: Some(true) }).await; outcome.expect("stop and publish video again timed out"); } + +/// An audio track that B drops while A is muted arrives again after the unmute +/// +/// The SFU sends the unmuted audio on the receiver that B already has, so +/// webrtc-rs fires no new `on_track`. The SDK delivers a new `RemoteTrack` for +/// that receiver when the audio arrives. +#[tokio::test] +async fn an_audio_track_dropped_while_muted_arrives_again_after_unmute() { + let Some(client) = common::client_or_skip() else { + return; + }; + init_tracing(); + + let user_a = common::unique_id("a"); + let user_b = common::unique_id("b"); + let (admin, call_id) = setup_call(&client, &[&user_a, &user_b]).await; + + let outcome = tokio::time::timeout(Duration::from_secs(150), async { + let call_a = client.video().call("default", &call_id); + let call_b = client.video().call("default", &call_id); + + let mut tracks_b = track_sink(&call_b); + call_b + .join(JoinCallData::new(&user_b)) + .await + .expect("B join"); + call_b + .update_subscriptions(SubscriptionConfig::audio_all()) + .await + .expect("B update_subscriptions"); + let mut events_b = call_b.sfu_events(); + + call_a + .join(JoinCallData::new(&user_a)) + .await + .expect("A join"); + let audio_a = LocalAudioTrack::opus().expect("opus track"); + call_a + .publish_audio(audio_a.clone()) + .await + .expect("A publish_audio"); + let feeder = spawn_tone(audio_a); + let remote_a = recv_track( + &mut tracks_b, + &user_a, + TrackType::Audio, + Duration::from_secs(45), + ) + .await + .expect("B never received A's audio track"); + + call_a.mute_track(TrackType::Audio).await.expect("A mute"); + assert!( + await_track_event( + &mut events_b, + &user_a, + TrackType::Audio, + false, + Duration::from_secs(45) + ) + .await, + "B never received A's audio TrackUnpublished after the mute" + ); + drop(remote_a); + call_a + .unmute_track(TrackType::Audio) + .await + .expect("A unmute"); + assert!( + await_track_event( + &mut events_b, + &user_a, + TrackType::Audio, + true, + Duration::from_secs(45) + ) + .await, + "B never received A's audio TrackPublished after the unmute" + ); + let again = recv_track( + &mut tracks_b, + &user_a, + TrackType::Audio, + Duration::from_secs(30), + ) + .await + .expect("B did not get A's audio track again after the unmute"); + let rms = drain_rms(&again, FRAME_20MS * 100, Duration::from_secs(30)).await; + feeder.abort(); + assert!( + rms > NON_SILENT_RMS, + "B got no audio on the track delivered again (rms={rms:.4})" + ); + + call_a.leave().await.expect("A leave"); + call_b.leave().await.expect("B leave"); + }) + .await; + + let _ = admin.delete(DeleteCallRequest { hard: Some(true) }).await; + outcome.expect("audio track dropped while muted timed out"); +} + +/// A video track that B drops while A is muted arrives again after the unmute +#[tokio::test] +async fn a_video_track_dropped_while_muted_arrives_again_after_unmute() { + let Some(client) = common::client_or_skip() else { + return; + }; + init_tracing(); + + let user_a = common::unique_id("a"); + let user_b = common::unique_id("b"); + let (admin, call_id) = setup_call(&client, &[&user_a, &user_b]).await; + + let outcome = tokio::time::timeout(Duration::from_secs(150), async { + let call_a = client.video().call("default", &call_id); + let call_b = client.video().call("default", &call_id); + + let mut tracks_b = track_sink(&call_b); + call_b + .join(JoinCallData::new(&user_b)) + .await + .expect("B join"); + call_b + .update_subscriptions(SubscriptionConfig::audio_video()) + .await + .expect("B update_subscriptions"); + let mut events_b = call_b.sfu_events(); + + call_a + .join(JoinCallData::new(&user_a)) + .await + .expect("A join"); + let video_a = LocalVideoTrack::vp9().expect("vp9 track"); + call_a + .publish_video(video_a.clone()) + .await + .expect("A publish_video"); + let feeder = spawn_blue_video(video_a); + let remote_a = recv_track( + &mut tracks_b, + &user_a, + TrackType::Video, + Duration::from_secs(45), + ) + .await + .expect("B never received A's video track"); + + call_a.mute_track(TrackType::Video).await.expect("A mute"); + assert!( + await_track_event( + &mut events_b, + &user_a, + TrackType::Video, + false, + Duration::from_secs(45) + ) + .await, + "B never received A's video TrackUnpublished after the mute" + ); + drop(remote_a); + call_a + .unmute_track(TrackType::Video) + .await + .expect("A unmute"); + assert!( + await_track_event( + &mut events_b, + &user_a, + TrackType::Video, + true, + Duration::from_secs(45) + ) + .await, + "B never received A's video TrackPublished after the unmute" + ); + let again = recv_track( + &mut tracks_b, + &user_a, + TrackType::Video, + Duration::from_secs(30), + ) + .await + .expect("B did not get A's video track again after the unmute"); + let frame = tokio::time::timeout(Duration::from_secs(45), again.next_video_frame()) + .await + .expect("B decoded no frame on the track delivered again") + .expect("the video track delivered again ended"); + feeder.abort(); + assert_packed_blue_frame(&frame); + + call_a.leave().await.expect("A leave"); + call_b.leave().await.expect("B leave"); + }) + .await; + + let _ = admin.delete(DeleteCallRequest { hard: Some(true) }).await; + outcome.expect("video track dropped while muted timed out"); +} + +/// A track that B keeps through a mute gets the audio again after the unmute, +/// and B gets no second track for it +#[tokio::test] +async fn a_kept_track_resumes_after_unmute_without_a_second_track() { + let Some(client) = common::client_or_skip() else { + return; + }; + init_tracing(); + + let user_a = common::unique_id("a"); + let user_b = common::unique_id("b"); + let (admin, call_id) = setup_call(&client, &[&user_a, &user_b]).await; + + let outcome = tokio::time::timeout(Duration::from_secs(150), async { + let call_a = client.video().call("default", &call_id); + let call_b = client.video().call("default", &call_id); + + let mut tracks_b = track_sink(&call_b); + call_b + .join(JoinCallData::new(&user_b)) + .await + .expect("B join"); + call_b + .update_subscriptions(SubscriptionConfig::audio_all()) + .await + .expect("B update_subscriptions"); + let mut events_b = call_b.sfu_events(); + + call_a + .join(JoinCallData::new(&user_a)) + .await + .expect("A join"); + let audio_a = LocalAudioTrack::opus().expect("opus track"); + call_a + .publish_audio(audio_a.clone()) + .await + .expect("A publish_audio"); + let feeder = spawn_tone(audio_a); + let remote_a = recv_track( + &mut tracks_b, + &user_a, + TrackType::Audio, + Duration::from_secs(45), + ) + .await + .expect("B never received A's audio track"); + + call_a.mute_track(TrackType::Audio).await.expect("A mute"); + assert!( + await_track_event( + &mut events_b, + &user_a, + TrackType::Audio, + false, + Duration::from_secs(45) + ) + .await, + "B never received A's audio TrackUnpublished after the mute" + ); + call_a + .unmute_track(TrackType::Audio) + .await + .expect("A unmute"); + assert!( + await_track_event( + &mut events_b, + &user_a, + TrackType::Audio, + true, + Duration::from_secs(45) + ) + .await, + "B never received A's audio TrackPublished after the unmute" + ); + let rms = drain_rms(&remote_a, FRAME_20MS * 100, Duration::from_secs(30)).await; + assert!( + rms > NON_SILENT_RMS, + "B got no audio on its kept track after the unmute (rms={rms:.4})" + ); + assert!( + recv_track( + &mut tracks_b, + &user_a, + TrackType::Audio, + Duration::from_secs(3) + ) + .await + .is_none(), + "B got A's audio track a second time" + ); + feeder.abort(); + + call_a.leave().await.expect("A leave"); + call_b.leave().await.expect("B leave"); + }) + .await; + + let _ = admin.delete(DeleteCallRequest { hard: Some(true) }).await; + outcome.expect("kept track through a mute timed out"); +} + +/// A track that B drops while A keeps sending does not come back +#[tokio::test] +async fn a_dropped_live_track_does_not_come_back() { + let Some(client) = common::client_or_skip() else { + return; + }; + init_tracing(); + + let user_a = common::unique_id("a"); + let user_b = common::unique_id("b"); + let (admin, call_id) = setup_call(&client, &[&user_a, &user_b]).await; + + let outcome = tokio::time::timeout(Duration::from_secs(150), async { + let call_a = client.video().call("default", &call_id); + let call_b = client.video().call("default", &call_id); + + let mut tracks_b = track_sink(&call_b); + call_b + .join(JoinCallData::new(&user_b)) + .await + .expect("B join"); + call_b + .update_subscriptions(SubscriptionConfig::audio_all()) + .await + .expect("B update_subscriptions"); + + call_a + .join(JoinCallData::new(&user_a)) + .await + .expect("A join"); + let audio_a = LocalAudioTrack::opus().expect("opus track"); + call_a + .publish_audio(audio_a.clone()) + .await + .expect("A publish_audio"); + let feeder = spawn_tone(audio_a); + let remote_a = recv_track( + &mut tracks_b, + &user_a, + TrackType::Audio, + Duration::from_secs(45), + ) + .await + .expect("B never received A's audio track"); + let rms = drain_rms(&remote_a, FRAME_20MS * 25, Duration::from_secs(30)).await; + assert!(rms > NON_SILENT_RMS, "B got no audio before the drop"); + + drop(remote_a); + assert!( + recv_track( + &mut tracks_b, + &user_a, + TrackType::Audio, + Duration::from_secs(5) + ) + .await + .is_none(), + "B got A's audio track again after it dropped the track" + ); + feeder.abort(); + + call_a.leave().await.expect("A leave"); + call_b.leave().await.expect("B leave"); + }) + .await; + + let _ = admin.delete(DeleteCallRequest { hard: Some(true) }).await; + outcome.expect("dropped live track timed out"); +}