From a2215b073c17ef8aec9250688f99f1d5371acc5d Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 5 Oct 2026 12:40:28 +0200 Subject: [PATCH 1/6] fix: keep pacing published audio through publisher disconnects --- examples/gpt_realtime_bot.rs | 2 -- src/rtc/join/connection.rs | 11 ++++--- src/rtc/join/publish.rs | 15 ++++----- src/rtc/join/reconnect_runtime.rs | 2 +- src/rtc/join/tests.rs | 51 +++++++++++++++++-------------- src/rtc/peer/publisher.rs | 8 +---- src/rtc/tracks/local.rs | 33 +++++++------------- 7 files changed, 55 insertions(+), 67 deletions(-) diff --git a/examples/gpt_realtime_bot.rs b/examples/gpt_realtime_bot.rs index c6019eb..707e28c 100644 --- a/examples/gpt_realtime_bot.rs +++ b/examples/gpt_realtime_bot.rs @@ -334,8 +334,6 @@ async fn configure_openai( Box::pin(async move { if state == RTCPeerConnectionState::Connected { paced_mic.start_pacing().await; - } else { - paced_mic.pause_pacing(); } }) })); diff --git a/src/rtc/join/connection.rs b/src/rtc/join/connection.rs index 680dcf4..5733af4 100644 --- a/src/rtc/join/connection.rs +++ b/src/rtc/join/connection.rs @@ -166,12 +166,13 @@ pub(super) fn register_connection_state( tracing::debug!(label, ?state, "stream.rtc.pc.state"); tracer.trace("connectionstatechange", json!(state.to_string())); // Not awaited: waiting for the media lock here can block ICE - // state delivery. The sync reads the current publisher state, so - // a subscriber change applies the same state again. - if core.is_generation_current(generation) { - let sync_core = core.clone(); + // state delivery. The task reads the current publisher state, so + // a subscriber change only repeats the check. + if state == RTCPeerConnectionState::Connected && core.is_generation_current(generation) + { + let pacing_core = core.clone(); std::mem::drop(core.spawn_generation_task(generation, async move { - sync_core.sync_audio_pacing().await; + pacing_core.start_audio_pacing_if_connected().await; })); } if state == RTCPeerConnectionState::Connected { diff --git a/src/rtc/join/publish.rs b/src/rtc/join/publish.rs index 94cdd04..476bbe2 100644 --- a/src/rtc/join/publish.rs +++ b/src/rtc/join/publish.rs @@ -166,8 +166,10 @@ impl RtcCore { Ok(()) } - /// Pace the published audio only while the current publisher is connected. - pub(super) async fn sync_audio_pacing(&self) { + /// Start pacing the published audio once the current publisher connects. + /// Pacing continues through later disconnects, so a reconnect adds no + /// delay; stream-py `AudioStreamTrack` also catches up after a gap. + pub(super) async fn start_audio_pacing_if_connected(&self) { let media = self.media.lock().await; let connected = self .publisher_handles() @@ -175,12 +177,11 @@ impl RtcCore { .is_some_and(|(publisher, ..)| { publisher.connection_state() == RTCPeerConnectionState::Connected }); + if !connected { + return; + } for track in media.active_tracks() { - if connected { - track.start_audio_pacing().await; - } else { - track.pause_audio_pacing(); - } + track.start_audio_pacing().await; } } diff --git a/src/rtc/join/reconnect_runtime.rs b/src/rtc/join/reconnect_runtime.rs index 74773aa..7dc7c61 100644 --- a/src/rtc/join/reconnect_runtime.rs +++ b/src/rtc/join/reconnect_runtime.rs @@ -5,7 +5,7 @@ use super::*; impl RtcCore { /// Re-publish every local track on the fresh publisher PC after a /// REJOIN/MIGRATE. The tracks are the same `Arc`-backed handles, so they bind - /// to the new PeerConnection and pacing resumes when it connects. + /// to the new PeerConnection and their pacing continues. pub(super) async fn restore_published_tracks(&self) -> Result<()> { let mut media = self.media.lock().await; let capabilities = self diff --git a/src/rtc/join/tests.rs b/src/rtc/join/tests.rs index ecf616b..4c70102 100644 --- a/src/rtc/join/tests.rs +++ b/src/rtc/join/tests.rs @@ -984,23 +984,13 @@ async fn stale_coordinator_stop_keeps_the_current_coordinator() { } #[tokio::test] -async fn published_audio_is_paced_only_while_the_publisher_is_connected() { +async fn published_audio_paces_from_the_first_connect_through_a_disconnect() { let core = test_core(); let generation = prepare_joined_core(&core, "alice"); let (connection, _sfu) = establish_fake(&core, generation).await; let publisher = connection.publisher.clone(); *core.connection.lock().await = Some(connection); let audio = LocalAudioTrack::opus().expect("opus track"); - let tone = (0..960_u32) - .map(|n| { - (12_000.0 * (std::f64::consts::TAU * 440.0 * f64::from(n) / 48_000.0).sin()) as i16 - }) - .collect::>() - .repeat(50); - audio - .write_pcm(PcmFrame::mono(tone, 48_000)) - .await - .expect("one second fits the queue"); publisher .add_track(audio.webrtc_track()) .await @@ -1009,21 +999,36 @@ async fn published_audio_is_paced_only_while_the_publisher_is_connected() { .lock() .await .begin_publish(LocalTrack::Audio(audio.clone()), 0); - - let (receiver, mut remote) = peer::connect_audio_receiver(&publisher).await; - - tokio::time::timeout(Duration::from_secs(5), remote.recv()) + // A second peer on the same track receives every paced packet. + let observer = peer::new_peer_connection(&[]).await.expect("observer"); + observer + .add_track(audio.webrtc_track()) + .await + .expect("add observed track"); + let (observer_receiver, mut observed) = peer::connect_audio_receiver(&observer).await; + let (receiver, _remote) = peer::connect_audio_receiver(&publisher).await; + let observed = tokio::time::timeout(Duration::from_secs(5), observed.recv()) .await - .expect("paced audio reaches the receiver") - .expect("remote track"); + .expect("pacing starts when the publisher connects") + .expect("observed track"); + publisher.close().await.expect("close publisher"); - wait_for( - Duration::from_secs(2), - || !audio.is_pacing(), - "closed publisher pauses pacing", - ) - .await; + let (at_close, _) = observed.read_rtp().await.expect("observed packet"); + let second_later = at_close.header.timestamp.wrapping_add(48_000); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let (packet, _) = observed.read_rtp().await.expect("observed packet"); + if packet.header.timestamp.wrapping_sub(second_later) < u32::MAX / 2 { + break; + } + } + }) + .await + .expect("pacing continues after the publisher closes"); + let _ = receiver.close().await; + let _ = observer_receiver.close().await; + let _ = observer.close().await; core.leave("test cleanup").await.expect("leave"); } diff --git a/src/rtc/peer/publisher.rs b/src/rtc/peer/publisher.rs index 25c48a2..e1435b6 100644 --- a/src/rtc/peer/publisher.rs +++ b/src/rtc/peer/publisher.rs @@ -101,11 +101,6 @@ pub(crate) async fn restart_ice( } } } - // Media is lost until ICE connects again; the publisher `Connected` state - // resumes pacing. - for track in tracks { - track.pause_audio_pacing(); - } negotiate_publish(publisher, signal, session_id, tracks, publish_options).await } @@ -385,13 +380,12 @@ mod tests { .expect("remote track") .expect("remote track channel"); let (mut last, _) = remote.read_rtp().await.expect("first packet"); - retired.pause_audio_pacing(); + retired.stop(); while let Ok(Ok((packet, _))) = tokio::time::timeout(Duration::from_millis(200), remote.read_rtp()).await { last = packet; } - retired.stop(); let track = LocalTrack::Audio(LocalAudioTrack::opus().expect("second track")); assert!( diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index 7cfd2d5..a21afca 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -7,9 +7,10 @@ //! - [`LocalAudioTrack::write_pcm`] — raw [`PcmFrame`]s. Resampled to 48 kHz //! mono, queued, and paced into 20 ms Opus frames by a background task that //! emits silence on starve (stream-py `AudioStreamTrack` pacing). Pacing runs -//! after [`LocalAudioTrack::start_pacing`]; a published track paces only -//! while the SFU publisher is connected. This is the PCM republish / TTS-bot -//! path. +//! after [`LocalAudioTrack::start_pacing`]; a published track starts pacing +//! when the SFU publisher first connects and does not pause on a later +//! disconnect, so audio written during an outage is lost instead of delayed. +//! This is the PCM republish / TTS-bot path. //! - [`LocalAudioTrack::write_sample`] / [`LocalVideoTrack::write_sample`] — //! already-encoded media (Opus/VP8/…) plus a frame duration; the SDK //! packetizes and writes. The caller controls pacing. @@ -600,26 +601,23 @@ impl LocalAudioTrack { /// the queue is empty. Before this, [`write_pcm`](Self::write_pcm) only /// fills the queue. /// - /// [`Call::publish_audio`](crate::Call::publish_audio) starts and pauses - /// pacing with the SFU publisher connection. Call this yourself only for a - /// track on your own PeerConnection (see [`webrtc_track`](Self::webrtc_track)), - /// after that PeerConnection connects. + /// [`Call::publish_audio`](crate::Call::publish_audio) starts pacing when + /// the SFU publisher first connects, and pacing continues through later + /// disconnects. Call this yourself only for a track on your own + /// PeerConnection (see [`webrtc_track`](Self::webrtc_track)), after that + /// PeerConnection connects. pub async fn start_pacing(&self) { self.inner.pacing_enabled.store(true, Ordering::SeqCst); self.ensure_pacer(); } - /// Stop taking queued PCM. The queue keeps its audio for the next + /// Stop taking queued PCM. The queue keeps its audio, so audio written + /// while paused is sent late after the next /// [`start_pacing`](Self::start_pacing). pub fn pause_pacing(&self) { self.inner.pacing_enabled.store(false, Ordering::SeqCst); } - #[cfg(test)] - pub(crate) fn is_pacing(&self) -> bool { - self.inner.pacing_enabled.load(Ordering::SeqCst) - } - /// Continue the RTP sequence numbers and timestamps of `previous`, whose /// sender this track takes over. The SFU drops a stream whose timestamps /// go back. @@ -2011,15 +2009,6 @@ impl LocalTrack { } } - pub(crate) fn pause_audio_pacing(&self) { - match self { - LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track) => { - track.pause_pacing(); - } - LocalTrack::Video { .. } => {} - } - } - pub(crate) fn stop(&self) { match self { LocalTrack::Audio(a) | LocalTrack::ScreenShareAudio(a) => a.stop(), From 28c4de33d6838a17458bc7a7be59b5823aa94ad3 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 5 Oct 2026 12:40:34 +0200 Subject: [PATCH 2/6] feat: add LocalAudioTrackConfig::pace and hold new publications until the SFU connects --- CHANGELOG.md | 16 +++++ examples/gpt_realtime_bot.rs | 8 ++- src/rtc/join/publication.rs | 2 + src/rtc/join/publish.rs | 7 +- src/rtc/join/tests.rs | 55 +++++++++++++++ src/rtc/peer/publisher.rs | 10 +-- src/rtc/tracks/local.rs | 133 +++++++++++++++++++++++++++++------ 7 files changed, 201 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 18826bd..82e9012 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -74,8 +74,24 @@ nothing (it was audio). write to a stopped track returns `RtcError::IllegalState`, so a later join must publish new tracks. +### The PCM queue holds up to 60 s + +`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. + ## New Features +### Pacing control for PCM tracks + +`LocalAudioTrackConfig::pace` (default `true`) starts pacing at the first +`write_pcm`, as before. With `with_pace(false)`, `write_pcm` only queues until +`LocalAudioTrack::start_pacing`; `pause_pacing` stops pacing and keeps the +queue. `Call::publish_audio` holds the queue of a new publication until the SFU +publisher first connects, so the audio written before the connect is not lost. +Pacing does not pause on a later disconnect: audio written during an outage is +lost, and a reconnect adds no delay. + ### A token-only client can prepare a call before the join `RtcClient::call` returns an `RtcCall` that is not joined yet, and diff --git a/examples/gpt_realtime_bot.rs b/examples/gpt_realtime_bot.rs index 707e28c..3ff4d7b 100644 --- a/examples/gpt_realtime_bot.rs +++ b/examples/gpt_realtime_bot.rs @@ -51,8 +51,8 @@ use anyhow::{Context, Result, anyhow, bail}; use getstream::models::{CallRequest, GetOrCreateCallRequest, MemberRequest, UserRequest}; use getstream::rtc::proto::models::TrackType; use getstream::rtc::{ - JoinCallData, LocalAudioTrack, LocalVideoTrack, RemoteTrack, RtcError, SubscriptionConfig, - TrackSubscriptionConfig, VideoFrame, + JoinCallData, LocalAudioTrack, LocalAudioTrackConfig, LocalVideoTrack, RemoteTrack, RtcError, + SubscriptionConfig, TrackSubscriptionConfig, VideoFrame, }; use getstream::video::Call; use getstream::{Stream, TokenOptions}; @@ -321,7 +321,9 @@ async fn configure_openai( .await .context("create oai-events data channel")?; - let mic = LocalAudioTrack::opus().context("opus track for OpenAI")?; + // Paced from the OpenAI connect below, so audio written before it is kept. + let mic = LocalAudioTrack::opus_with_config(LocalAudioTrackConfig::default().with_pace(false)) + .context("opus track for OpenAI")?; spawn_rtcp_drain( &tasks, pc.add_track(mic.webrtc_track()) diff --git a/src/rtc/join/publication.rs b/src/rtc/join/publication.rs index 9608fbe..9df6bc6 100644 --- a/src/rtc/join/publication.rs +++ b/src/rtc/join/publication.rs @@ -63,6 +63,7 @@ impl MediaState { if let Some(position) = self.position(&track_id) { return self.publications[position].status; } + track.hold_audio_pacing(); self.publications.push(LocalPublication { track, status: PublicationStatus::PendingPublishNegotiation, @@ -80,6 +81,7 @@ impl MediaState { pub(super) fn remove(&mut self, track_id: &str) -> Option { let position = self.position(track_id)?; let publication = self.publications.remove(position); + publication.track.release_audio_pacing(); let key = ( publication.publish_option_id, publication.track.track_type() as i32, diff --git a/src/rtc/join/publish.rs b/src/rtc/join/publish.rs index 476bbe2..d3a2530 100644 --- a/src/rtc/join/publish.rs +++ b/src/rtc/join/publish.rs @@ -166,9 +166,10 @@ impl RtcCore { Ok(()) } - /// Start pacing the published audio once the current publisher connects. - /// Pacing continues through later disconnects, so a reconnect adds no - /// delay; stream-py `AudioStreamTrack` also catches up after a gap. + /// End the hold of each new audio publication once the current publisher + /// connects, and start its pacing. Pacing continues through later + /// disconnects, so a reconnect adds no delay; stream-py `AudioStreamTrack` + /// also catches up after a gap. pub(super) async fn start_audio_pacing_if_connected(&self) { let media = self.media.lock().await; let connected = self diff --git a/src/rtc/join/tests.rs b/src/rtc/join/tests.rs index 4c70102..2e31cdd 100644 --- a/src/rtc/join/tests.rs +++ b/src/rtc/join/tests.rs @@ -1006,6 +1006,16 @@ async fn published_audio_paces_from_the_first_connect_through_a_disconnect() { .await .expect("add observed track"); let (observer_receiver, mut observed) = peer::connect_audio_receiver(&observer).await; + audio + .write_pcm(PcmFrame::mono(vec![0; 960], 48_000)) + .await + .expect("write before the publisher connects"); + assert!( + tokio::time::timeout(Duration::from_millis(300), observed.recv()) + .await + .is_err(), + "a publication takes no audio before the publisher connects" + ); let (receiver, _remote) = peer::connect_audio_receiver(&publisher).await; let observed = tokio::time::timeout(Duration::from_secs(5), observed.recv()) .await @@ -1032,6 +1042,51 @@ async fn published_audio_paces_from_the_first_connect_through_a_disconnect() { core.leave("test cleanup").await.expect("leave"); } +#[tokio::test] +async fn a_failed_publish_does_not_hold_the_track() { + let core = test_core(); + let generation = prepare_joined_core(&core, "alice"); + let (mut connection, _sfu) = establish_fake(&core, generation).await; + connection.publish_options = vec![models::PublishOption { + track_type: TrackType::Audio as i32, + codec: Some(models::Codec { + name: "opus".to_owned(), + ..Default::default() + }), + ..Default::default() + }]; + *core.connection.lock().await = Some(connection); + core.own_capabilities + .lock() + .unwrap_or_else(|error| error.into_inner()) + .insert("send-audio".to_owned()); + let audio = LocalAudioTrack::opus().expect("opus track"); + + // The fake SFU has no SetPublisher endpoint. + core.publish(LocalTrack::Audio(audio.clone())) + .await + .expect_err("publish without SetPublisher"); + + let sender = peer::new_peer_connection(&[]).await.expect("sender"); + sender + .add_track(audio.webrtc_track()) + .await + .expect("add track"); + let (receiver, mut remote) = peer::connect_audio_receiver(&sender).await; + audio + .write_pcm(PcmFrame::mono(vec![0; 960], 48_000)) + .await + .expect("write"); + tokio::time::timeout(Duration::from_secs(5), remote.recv()) + .await + .expect("the track paces on its own PeerConnection") + .expect("remote track"); + audio.stop(); + let _ = sender.close().await; + let _ = receiver.close().await; + core.leave("test cleanup").await.expect("leave"); +} + #[tokio::test] async fn only_the_latest_remote_track_unsubscribes_when_dropped_without_a_runtime() { let core = test_core(); diff --git a/src/rtc/peer/publisher.rs b/src/rtc/peer/publisher.rs index e1435b6..b496c29 100644 --- a/src/rtc/peer/publisher.rs +++ b/src/rtc/peer/publisher.rs @@ -368,13 +368,14 @@ mod tests { }), ..Default::default() }]; - let retired = LocalTrack::Audio(LocalAudioTrack::opus().expect("first track")); + let first = LocalAudioTrack::opus().expect("first track"); + let retired = LocalTrack::Audio(first.clone()); let publisher = peer::new_peer_connection(&[]).await.expect("publisher"); let rtcp_tasks = add_transceiver_for_track(&publisher, &retired, &opus) .await .expect("first transceiver"); let (receiver, mut remote_rx) = peer::connect_audio_receiver(&publisher).await; - retired.start_audio_pacing().await; + first.start_pacing().await; let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) .await .expect("remote track") @@ -387,13 +388,14 @@ mod tests { last = packet; } - let track = LocalTrack::Audio(LocalAudioTrack::opus().expect("second track")); + let second = LocalAudioTrack::opus().expect("second track"); + let track = LocalTrack::Audio(second.clone()); assert!( replace_retired_track(&publisher, &retired, &track, &opus) .await .expect("replace the retired track") ); - track.start_audio_pacing().await; + second.start_pacing().await; let (next, _) = remote.read_rtp().await.expect("second packet"); assert_eq!( diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index a21afca..b74b91c 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -6,9 +6,10 @@ //! //! - [`LocalAudioTrack::write_pcm`] — raw [`PcmFrame`]s. Resampled to 48 kHz //! mono, queued, and paced into 20 ms Opus frames by a background task that -//! emits silence on starve (stream-py `AudioStreamTrack` pacing). Pacing runs -//! after [`LocalAudioTrack::start_pacing`]; a published track starts pacing -//! when the SFU publisher first connects and does not pause on a later +//! emits silence on starve (stream-py `AudioStreamTrack` pacing). Pacing +//! starts at the first write, or at [`LocalAudioTrack::start_pacing`] when +//! [`LocalAudioTrackConfig::pace`] is off. A published track holds its queue +//! until the SFU publisher first connects and does not pause on a later //! disconnect, so audio written during an outage is lost instead of delayed. //! This is the PCM republish / TTS-bot path. //! - [`LocalAudioTrack::write_sample`] / [`LocalVideoTrack::write_sample`] — @@ -335,6 +336,11 @@ struct AudioInner { /// Set by `start_pacing`, cleared by `pause_pacing`. pacing_enabled: AtomicBool, pcm_pacing: AtomicBool, + /// [`LocalAudioTrackConfig::pace`]. + pace: bool, + /// Set while a new publication waits for the first SFU publisher connect. + /// The pacer takes no PCM while it is set. + held: AtomicBool, write_guard: tokio::sync::Mutex<()>, } @@ -356,6 +362,9 @@ pub struct LocalAudioTrackConfig { /// write above it drops the oldest queued samples. The minimum is one 20 ms /// frame. pub pcm_queue_capacity: Duration, + /// Start pacing at the first [`LocalAudioTrack::write_pcm`]. When `false`, + /// `write_pcm` only queues until [`LocalAudioTrack::start_pacing`]. + pub pace: bool, } impl Default for LocalAudioTrackConfig { @@ -366,6 +375,7 @@ impl Default for LocalAudioTrackConfig { expected_packet_loss_pct: EXPECTED_PACKET_LOSS_PCT, dtx: true, pcm_queue_capacity: PCM_QUEUE_CAPACITY, + pace: true, } } } @@ -407,6 +417,13 @@ impl LocalAudioTrackConfig { self.pcm_queue_capacity = pcm_queue_capacity; self } + + /// Enable or disable the pacing start at the first `write_pcm`. + #[must_use] + pub fn with_pace(mut self, pace: bool) -> Self { + self.pace = pace; + self + } } /// An outbound Opus audio track. @@ -470,6 +487,8 @@ impl LocalAudioTrack { pacer_started: AtomicBool::new(false), pacing_enabled: AtomicBool::new(false), pcm_pacing: AtomicBool::new(true), + pace: config.pace, + held: AtomicBool::new(false), write_guard: tokio::sync::Mutex::new(()), }), }) @@ -478,9 +497,11 @@ impl LocalAudioTrack { /// Queue a PCM frame for the paced 20 ms Opus encoder. /// /// The frame is resampled to 48 kHz mono and buffered up to - /// [`LocalAudioTrackConfig::pcm_queue_capacity`]. While pacing runs (see - /// [`start_pacing`](Self::start_pacing)), a background task emits one Opus - /// packet every 20 ms, writing silence when the buffer runs dry. This method + /// [`LocalAudioTrackConfig::pcm_queue_capacity`]. The first write starts + /// pacing when [`LocalAudioTrackConfig::pace`] is set (the default); + /// otherwise [`start_pacing`](Self::start_pacing) does. While pacing runs, a + /// background task emits one Opus packet every 20 ms, writing silence when + /// the buffer runs dry. This method /// does not backpressure a producer: overflow drops the oldest queued /// samples and retains the newest audio. [`flush`](Self::flush) still drops /// all unsent samples immediately for barge-in. @@ -518,6 +539,11 @@ impl LocalAudioTrack { } dropped }; + // Only a pacer that never started starts here, so a `pause_pacing` + // stays in force. + if self.inner.pace && !self.inner.pacer_started.load(Ordering::SeqCst) { + self.start_pacing().await; + } if dropped > 0 { Err(RtcError::PcmQueueOverflow { dropped_samples: dropped, @@ -598,14 +624,14 @@ impl LocalAudioTrack { } /// Start taking queued PCM: one 20 ms frame every 20 ms, or silence when - /// the queue is empty. Before this, [`write_pcm`](Self::write_pcm) only - /// fills the queue. + /// the queue is empty. With [`LocalAudioTrackConfig::pace`] set (the + /// default), the first [`write_pcm`](Self::write_pcm) calls this. /// - /// [`Call::publish_audio`](crate::Call::publish_audio) starts pacing when - /// the SFU publisher first connects, and pacing continues through later - /// disconnects. Call this yourself only for a track on your own - /// PeerConnection (see [`webrtc_track`](Self::webrtc_track)), after that - /// PeerConnection connects. + /// [`Call::publish_audio`](crate::Call::publish_audio) holds the queue + /// until the SFU publisher first connects, then starts pacing; pacing + /// continues through later disconnects. With `pace` off, call this for a + /// track on your own PeerConnection (see [`webrtc_track`](Self::webrtc_track)) + /// after that PeerConnection connects. pub async fn start_pacing(&self) { self.inner.pacing_enabled.store(true, Ordering::SeqCst); self.ensure_pacer(); @@ -618,6 +644,17 @@ impl LocalAudioTrack { self.inner.pacing_enabled.store(false, Ordering::SeqCst); } + /// Take no PCM until [`Self::release_pacing`]. A new publication holds its + /// track until the SFU publisher first connects. + pub(crate) fn hold_pacing(&self) { + self.inner.held.store(true, Ordering::SeqCst); + } + + /// End a [`Self::hold_pacing`]. Returns whether the track was held. + pub(crate) fn release_pacing(&self) -> bool { + self.inner.held.swap(false, Ordering::SeqCst) + } + /// Continue the RTP sequence numbers and timestamps of `previous`, whose /// sender this track takes over. The SFU drops a stream whose timestamps /// go back. @@ -657,8 +694,10 @@ impl LocalAudioTrack { /// [`Call::publish_audio`](crate::Call::publish_audio) does this for the /// SFU; you only need it to send the same audio to a second peer, such as an /// AI provider's Realtime endpoint. Every write path (`write_pcm` and - /// friends) feeds all bound senders. A track that is not published paces - /// `write_pcm` audio only after [`start_pacing`](Self::start_pacing). + /// friends) feeds all bound senders. With [`LocalAudioTrackConfig::pace`] + /// set, audio written before this PeerConnection connects is lost; turn it + /// off and call [`start_pacing`](Self::start_pacing) at the connect to keep + /// that audio. pub fn webrtc_track(&self) -> Arc { self.inner.core.track.clone() } @@ -703,7 +742,9 @@ async fn pace_audio(track: Weak) { if inner.core.stopped.load(Ordering::SeqCst) { return; } - if !inner.pcm_pacing.load(Ordering::SeqCst) || !inner.pacing_enabled.load(Ordering::SeqCst) + if !inner.pcm_pacing.load(Ordering::SeqCst) + || !inner.pacing_enabled.load(Ordering::SeqCst) + || inner.held.load(Ordering::SeqCst) { continue; } @@ -2000,10 +2041,32 @@ impl LocalTrack { } } + /// End the publication hold at the first SFU publisher connect and start + /// pacing. A track that is not held keeps its pacing state, so a + /// `pause_pacing` stays in force. pub(crate) async fn start_audio_pacing(&self) { match self { LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track) => { - track.start_pacing().await; + if track.release_pacing() { + track.start_pacing().await; + } + } + LocalTrack::Video { .. } => {} + } + } + + pub(crate) fn hold_audio_pacing(&self) { + match self { + LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track) => track.hold_pacing(), + LocalTrack::Video { .. } => {} + } + } + + /// End the hold of a publication that did not complete. + pub(crate) fn release_audio_pacing(&self) { + match self { + LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track) => { + track.release_pacing(); } LocalTrack::Video { .. } => {} } @@ -2245,7 +2308,9 @@ mod tests { #[tokio::test] async fn write_pcm_above_the_default_minute_keeps_the_newest_samples() { - let track = LocalAudioTrack::opus().expect("opus track"); + let track = + LocalAudioTrack::opus_with_config(LocalAudioTrackConfig::default().with_pace(false)) + .expect("opus track"); let minute = 60 * OPUS_SAMPLE_RATE as usize; track .write_pcm(PcmFrame::mono(vec![1; minute], OPUS_SAMPLE_RATE)) @@ -2325,10 +2390,36 @@ mod tests { assert!(with_capacity(Duration::from_millis(20)).is_ok()); } + #[tokio::test] + async fn the_first_write_starts_pacing_by_default() { + let track = LocalAudioTrack::opus().expect("opus track"); + let sender = peer::new_peer_connection(&[]).await.expect("sender"); + sender + .add_track(track.webrtc_track()) + .await + .expect("add track"); + let (receiver, mut remote_rx) = peer::connect_audio_receiver(&sender).await; + + track + .write_pcm(PcmFrame::mono(tone_20ms(), OPUS_SAMPLE_RATE)) + .await + .expect("write"); + + tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("paced audio reaches the receiver") + .expect("remote track channel"); + track.stop(); + let _ = sender.close().await; + let _ = receiver.close().await; + } + #[tokio::test] async fn pcm_written_before_pacing_starts_stays_queued() { let track = LocalAudioTrack::opus_with_config( - LocalAudioTrackConfig::default().with_pcm_queue_capacity(Duration::from_millis(100)), + LocalAudioTrackConfig::default() + .with_pcm_queue_capacity(Duration::from_millis(100)) + .with_pace(false), ) .expect("opus track"); track @@ -2380,7 +2471,9 @@ mod tests { #[tokio::test] async fn pcm_queued_before_the_connection_is_sent_from_its_first_sample() { - let track = LocalAudioTrack::opus().expect("opus track"); + let track = + LocalAudioTrack::opus_with_config(LocalAudioTrackConfig::default().with_pace(false)) + .expect("opus track"); let minute = tone_20ms().repeat(3_000); let queued = minute.len(); track From e8323b61b4395ecc4d1c6c2ec03dd19a89cc1e79 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 5 Oct 2026 13:25:51 +0200 Subject: [PATCH 3/6] feat: reuse the stopped video sender for a new video track --- CHANGELOG.md | 10 ++ src/rtc/client.rs | 5 +- src/rtc/error.rs | 13 +++ src/rtc/join/publication.rs | 36 ++++--- src/rtc/join/publish.rs | 33 ++++--- src/rtc/join/tests.rs | 63 ++++++++++++ src/rtc/peer/connection.rs | 14 ++- src/rtc/peer/mod.rs | 2 +- src/rtc/peer/publisher.rs | 191 +++++++++++++++++++++++++++++++++--- src/rtc/tracks/local.rs | 87 ++++++++++++++-- src/video/call.rs | 6 +- tests/rtc_media.rs | 126 ++++++++++++++++++++++++ 12 files changed, 533 insertions(+), 53 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 82e9012..91e3c39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -92,6 +92,16 @@ publisher first connects, so the audio written before the connect is not lost. Pacing does not pause on a later disconnect: audio written during an outage is lost, and a reconnect adds no delay. +### A stopped track can be replaced by a new track + +After `stop_publish`, a new audio, video, or screen-share track of the same kind +and publish option takes over the sender that the stop kept, as JS +`replaceTrack` does. Remote participants keep their remote track and get the +media of the new track. Before, a second video publish failed with +"participant not found". A simulcast track (`h264_simulcast`, `vp8_simulcast`) +cannot replace or be replaced: `publish` returns `RtcError::SimulcastReplace`. +Turn such a track off and on with `mute_track` and `unmute_track`. + ### A token-only client can prepare a call before the join `RtcClient::call` returns an `RtcCall` that is not joined yet, and diff --git a/src/rtc/client.rs b/src/rtc/client.rs index b2e16ee..7f75578 100644 --- a/src/rtc/client.rs +++ b/src/rtc/client.rs @@ -378,7 +378,10 @@ impl RtcCall { self.core.stop_noise_cancellation().await } - /// Stop publishing a local media track. + /// Stop publishing a local media track. A new track of the same kind that + /// is published later takes over its sender, except for a simulcast track + /// ([`RtcError::SimulcastReplace`]); use [`mute_track`](Self::mute_track) + /// for that. pub async fn stop_publish(&self, track: LocalTrack) -> Result<()> { self.core.stop_publish(track).await } diff --git a/src/rtc/error.rs b/src/rtc/error.rs index 1b966e7..29be991 100644 --- a/src/rtc/error.rs +++ b/src/rtc/error.rs @@ -132,6 +132,19 @@ pub enum RtcError { track_type: models::TrackType, }, + /// A stopped simulcast publication and a new track cannot share a sender: + /// webrtc-rs replaces the track only on a sender with one encoding. Turn a + /// simulcast track off and on with `mute_track` and `unmute_track` instead + /// of a stop. + #[error( + "a stopped {track_type:?} publication cannot change its track when either \ + track is simulcast; use mute_track and unmute_track" + )] + SimulcastReplace { + /// Publication kind requested by the caller. + track_type: models::TrackType, + }, + /// A PCM write exceeded the track's PCM queue capacity. The newest samples /// were retained and this many oldest samples were discarded. #[error( diff --git a/src/rtc/join/publication.rs b/src/rtc/join/publication.rs index 9df6bc6..b6870b0 100644 --- a/src/rtc/join/publication.rs +++ b/src/rtc/join/publication.rs @@ -27,7 +27,7 @@ pub(super) struct LocalPublication { pub(super) struct MediaState { pub(super) publications: Vec, pub(super) publish_quality: HashMap<(i32, i32), Vec>, - /// The publish option id and stopped track of each audio sender that + /// The publish option id and stopped track of each sender that /// `stop_publish` kept in the publisher envelope. retired: Vec<(i32, LocalTrack)>, } @@ -98,17 +98,11 @@ impl MediaState { } /// Remove the publication of a stopped track whose sender stays in the - /// publisher envelope. Keep an audio sender for a later publish; `publish` - /// never reuses a video sender. + /// publisher envelope, and keep that sender for a later publish. pub(super) fn retire(&mut self, track_id: &str) -> Option { let publish_option_id = self.publications[self.position(track_id)?].publish_option_id; let track = self.remove(track_id)?; - if matches!( - track.track_type(), - TrackType::Audio | TrackType::ScreenShareAudio - ) { - self.retired.push((publish_option_id, track.clone())); - } + self.retired.push((publish_option_id, track.clone())); Some(track) } @@ -120,19 +114,33 @@ impl MediaState { } } - /// Take the stopped track of the latest sender that [`Self::retire`] kept - /// for this kind of track. + /// The stopped track of the latest sender that [`Self::retire`] kept for + /// this kind of track. + pub(super) fn retired( + &self, + track_type: TrackType, + publish_option_id: i32, + ) -> Option<&LocalTrack> { + let position = self.retired_position(track_type, publish_option_id)?; + Some(&self.retired[position].1) + } + + /// Take the track that [`Self::retired`] returns. pub(super) fn take_retired( &mut self, track_type: TrackType, publish_option_id: i32, ) -> Option { - let position = self.retired.iter().rposition(|(option_id, retired)| { - retired.track_type() == track_type && *option_id == publish_option_id - })?; + let position = self.retired_position(track_type, publish_option_id)?; Some(self.retired.remove(position).1) } + fn retired_position(&self, track_type: TrackType, publish_option_id: i32) -> Option { + self.retired.iter().rposition(|(option_id, retired)| { + retired.track_type() == track_type && *option_id == publish_option_id + }) + } + pub(super) fn refresh_publish_options( &mut self, publish_options: &[models::PublishOption], diff --git a/src/rtc/join/publish.rs b/src/rtc/join/publish.rs index d3a2530..f837c17 100644 --- a/src/rtc/join/publish.rs +++ b/src/rtc/join/publish.rs @@ -63,25 +63,28 @@ impl RtcCore { }; let mut publisher_rtcp_tasks = Vec::new(); if status.is_none() { - // Reuse the audio sender that `stop_publish` kept, as JS `replaceTrack` - // does: a second audio m-line fails the SFU negotiation. Video RTCP - // readers keep the old track, and simulcast cannot be replaced. - let retired = if matches!( - track.track_type(), - TrackType::Audio | TrackType::ScreenShareAudio - ) { - media.take_retired(track.track_type(), publish_option_id) - } else { - None - }; - let reused = match retired { + // Reuse the sender that `stop_publish` kept, as JS `replaceTrack` + // does: a second m-line for the same publish option fails the SFU + // negotiation. + if media + .retired(track.track_type(), publish_option_id) + .is_some_and(|retired| retired.is_simulcast() || track.is_simulcast()) + { + return Err(RtcError::SimulcastReplace { + track_type: track.track_type(), + }); + } + let reader = match media.take_retired(track.track_type(), publish_option_id) { Some(retired) => { publisher::replace_retired_track(&publisher, &retired, &track, &publish_options) .await? } - None => false, + None => None, }; - if !reused { + let reused = reader.is_some(); + if let Some(reader) = reader { + self.register_publisher_tasks(vec![reader]).await; + } else { publisher_rtcp_tasks = match publisher::add_transceiver_for_track( &publisher, &track, @@ -208,6 +211,8 @@ impl RtcCore { /// (`Invalid SetPublisher request; ... new track must have the same envelope as /// previous`), so removing the sole sender and renegotiating an empty envelope /// fails on the wire; muting the track type is the wire-correct way to stop. + /// A later [`Self::publish`] of the same kind and publish option puts its + /// track on the kept sender. pub async fn stop_publish(self: &Arc, track: LocalTrack) -> Result<()> { let mut media = self.media.lock().await; let track_id = track.track_id(); diff --git a/src/rtc/join/tests.rs b/src/rtc/join/tests.rs index 2e31cdd..6810d35 100644 --- a/src/rtc/join/tests.rs +++ b/src/rtc/join/tests.rs @@ -1087,6 +1087,69 @@ async fn a_failed_publish_does_not_hold_the_track() { core.leave("test cleanup").await.expect("leave"); } +#[tokio::test] +async fn a_stopped_simulcast_publication_is_not_replaced() { + let core = test_core(); + let generation = prepare_joined_core(&core, "alice"); + let (mut connection, _sfu) = establish_fake(&core, generation).await; + connection.publish_options = vec![models::PublishOption { + id: 1, + track_type: TrackType::Video as i32, + codec: Some(models::Codec { + name: "h264".to_owned(), + ..Default::default() + }), + ..Default::default() + }]; + let stopped = LocalTrack::Video { + track: LocalVideoTrack::h264_simulcast().expect("simulcast track"), + track_type: TrackType::Video, + }; + let reader_tasks = crate::rtc::peer::publisher::add_transceiver_for_track( + &connection.publisher, + &stopped, + &connection.publish_options, + ) + .await + .expect("simulcast transceiver"); + *core.connection.lock().await = Some(connection); + core.own_capabilities + .lock() + .unwrap_or_else(|error| error.into_inner()) + .insert("send-video".to_owned()); + { + let mut media = core.media.lock().await; + media.begin_publish(stopped.clone(), 1); + media.retire(&stopped.track_id()); + } + stopped.stop(); + + for track in [ + LocalVideoTrack::h264_simulcast().expect("new simulcast track"), + LocalVideoTrack::h264().expect("new single-layer track"), + ] { + let result = core + .publish(LocalTrack::Video { + track, + track_type: TrackType::Video, + }) + .await; + assert!( + matches!( + result, + Err(RtcError::SimulcastReplace { + track_type: TrackType::Video + }) + ), + "{result:?}" + ); + } + for task in reader_tasks { + task.abort(); + } + core.leave("test cleanup").await.expect("leave"); +} + #[tokio::test] async fn only_the_latest_remote_track_unsubscribes_when_dropped_without_a_runtime() { let core = test_core(); diff --git a/src/rtc/peer/connection.rs b/src/rtc/peer/connection.rs index d3fda04..e7e9279 100644 --- a/src/rtc/peer/connection.rs +++ b/src/rtc/peer/connection.rs @@ -264,6 +264,18 @@ pub(crate) async fn connect_audio_receiver( ) -> ( Arc, tokio::sync::mpsc::Receiver>, +) { + connect_receiver(sender, RTPCodecType::Audio).await +} + +/// [`connect_audio_receiver`] for a receive-only peer of `kind`. +#[cfg(test)] +pub(crate) async fn connect_receiver( + sender: &RTCPeerConnection, + kind: RTPCodecType, +) -> ( + Arc, + tokio::sync::mpsc::Receiver>, ) { use std::time::Duration; use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState; @@ -274,7 +286,7 @@ pub(crate) async fn connect_audio_receiver( let receiver = new_peer_connection(&[]).await.expect("receiver"); receiver .add_transceiver_from_kind( - RTPCodecType::Audio, + kind, Some(RTCRtpTransceiverInit { direction: RTCRtpTransceiverDirection::Recvonly, send_encodings: vec![], diff --git a/src/rtc/peer/mod.rs b/src/rtc/peer/mod.rs index 70b4f30..5b7e079 100644 --- a/src/rtc/peer/mod.rs +++ b/src/rtc/peer/mod.rs @@ -13,7 +13,7 @@ pub mod publisher; mod subscriber; #[cfg(test)] -pub(crate) use connection::connect_audio_receiver; +pub(crate) use connection::{connect_audio_receiver, connect_receiver}; pub use connection::{generic_sdp, new_peer_connection, to_rtc_ice_servers, trace_peer_events}; pub(super) use ice::{PendingIce, register_ice_trickle}; pub(super) use subscriber::negotiate_subscriber; diff --git a/src/rtc/peer/publisher.rs b/src/rtc/peer/publisher.rs index b496c29..983b627 100644 --- a/src/rtc/peer/publisher.rs +++ b/src/rtc/peer/publisher.rs @@ -228,13 +228,14 @@ pub(crate) async fn add_transceiver_for_track( } /// Put `track` on the sender that still carries the stopped track `retired`, -/// as JS `replaceTrack` does. Returns `false` if `publisher` has no such sender. +/// as JS `replaceTrack` does. Returns the RTCP reader of `track`, or `None` if +/// `publisher` has no such sender. pub(crate) async fn replace_retired_track( publisher: &Arc, retired: &LocalTrack, track: &LocalTrack, publish_options: &[PublishOption], -) -> Result { +) -> Result>> { let retired_track_id = retired.track_id(); for transceiver in publisher.get_transceivers().await { let sender = transceiver.sender().await; @@ -251,13 +252,14 @@ pub(crate) async fn replace_retired_track( let physical = track.webrtc_tracks().into_iter().next().ok_or_else(|| { RtcError::Media("local publication has no physical encodings".to_owned()) })?; + let rid = physical.rid().map(str::to_owned); sender .replace_track(Some(physical)) .await .map_err(RtcError::from)?; - return Ok(true); + return Ok(Some(spawn_rtcp_reader(sender, rid, track.clone()))); } - Ok(false) + Ok(None) } fn spawn_rtcp_reader( @@ -267,12 +269,20 @@ fn spawn_rtcp_reader( ) -> JoinHandle<()> { tokio::spawn(async move { loop { - let result = match rid.as_deref() { - Some(rid) => { - let mut buffer = vec![0_u8; 1_500]; - sender.read_simulcast(&mut buffer, rid).await + let read = async { + match rid.as_deref() { + Some(rid) => { + let mut buffer = vec![0_u8; 1_500]; + sender.read_simulcast(&mut buffer, rid).await + } + None => sender.read_rtcp().await, } - None => sender.read_rtcp().await, + }; + let result = tokio::select! { + biased; + // A track that takes over this sender starts its own reader. + () = track.stopped() => break, + result = read => result, }; let Ok((packets, _attributes)) = result else { break; @@ -357,6 +367,7 @@ mod tests { use crate::rtc::proto::models::{Codec, VideoDimension}; use crate::rtc::tracks::{LocalAudioTrack, LocalVideoTrack, LocalVideoTrackConfig}; use std::time::Duration; + use webrtc::rtp_transceiver::rtp_codec::RTPCodecType; #[tokio::test] async fn a_track_on_a_retired_sender_continues_its_rtp_timeline() { @@ -390,11 +401,10 @@ mod tests { let second = LocalAudioTrack::opus().expect("second track"); let track = LocalTrack::Audio(second.clone()); - assert!( - replace_retired_track(&publisher, &retired, &track, &opus) - .await - .expect("replace the retired track") - ); + let reader = replace_retired_track(&publisher, &retired, &track, &opus) + .await + .expect("replace the retired track") + .expect("the retired sender"); second.start_pacing().await; let (next, _) = remote.read_rtp().await.expect("second packet"); @@ -407,6 +417,159 @@ mod tests { last.header.timestamp.wrapping_add(960) ); track.stop(); + reader.abort(); + for task in rtcp_tasks { + task.abort(); + } + let _ = publisher.close().await; + let _ = receiver.close().await; + } + + #[tokio::test] + async fn a_video_track_on_a_retired_sender_continues_its_rtp_timeline() { + let vp9 = [video_option("VP9")]; + let first = LocalVideoTrack::vp9().expect("first track"); + let retired = LocalTrack::Video { + track: first.clone(), + track_type: TrackType::Video, + }; + let publisher = peer::new_peer_connection(&[]).await.expect("publisher"); + let rtcp_tasks = add_transceiver_for_track(&publisher, &retired, &vp9) + .await + .expect("first transceiver"); + let (receiver, mut remote_rx) = + peer::connect_receiver(&publisher, RTPCodecType::Video).await; + let frame = vec![128_u8; 320 * 240 * 3 / 2]; + let frame_time = Duration::from_millis(33); + first + .write_i420(&frame, 320, 240, frame_time) + .await + .expect("first frame"); + let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + for _ in 0..3 { + first + .write_i420(&frame, 320, 240, frame_time) + .await + .expect("frame"); + } + retired.stop(); + let mut last = None; + while let Ok(Ok((packet, _))) = + tokio::time::timeout(Duration::from_millis(200), remote.read_rtp()).await + { + last = Some(packet); + } + let last = last.expect("packets of the first track"); + + let second = LocalVideoTrack::vp9().expect("second track"); + let track = LocalTrack::Video { + track: second.clone(), + track_type: TrackType::Video, + }; + let reader = replace_retired_track(&publisher, &retired, &track, &vp9) + .await + .expect("replace the retired track") + .expect("the retired sender"); + second + .write_i420(&frame, 320, 240, frame_time) + .await + .expect("second track frame"); + let (next, _) = tokio::time::timeout(Duration::from_secs(5), remote.read_rtp()) + .await + .expect("the receiver takes the packets of the second track") + .expect("second track packet"); + + assert_eq!( + next.header.sequence_number, + last.header.sequence_number.wrapping_add(1) + ); + assert_eq!( + next.header.timestamp, + last.header.timestamp.wrapping_add(2_970) + ); + assert!( + tokio::time::timeout(Duration::from_millis(200), remote_rx.recv()) + .await + .is_err(), + "the receiver keeps its remote track" + ); + track.stop(); + reader.abort(); + for task in rtcp_tasks { + task.abort(); + } + let _ = publisher.close().await; + let _ = receiver.close().await; + } + + #[tokio::test] + async fn a_keyframe_request_reaches_the_track_on_a_retired_sender() { + use webrtc::rtcp::payload_feedbacks::picture_loss_indication::PictureLossIndication; + + let vp9 = [video_option("VP9")]; + let retired = LocalTrack::Video { + track: LocalVideoTrack::vp9().expect("first track"), + track_type: TrackType::Video, + }; + let publisher = peer::new_peer_connection(&[]).await.expect("publisher"); + let rtcp_tasks = add_transceiver_for_track(&publisher, &retired, &vp9) + .await + .expect("first transceiver"); + let (receiver, mut remote_rx) = + peer::connect_receiver(&publisher, RTPCodecType::Video).await; + retired.stop(); + let second = LocalVideoTrack::vp9().expect("second track"); + let track = LocalTrack::Video { + track: second.clone(), + track_type: TrackType::Video, + }; + let reader = replace_retired_track(&publisher, &retired, &track, &vp9) + .await + .expect("replace the retired track") + .expect("the retired sender"); + let frame = vec![128_u8; 320 * 240 * 3 / 2]; + let frame_time = Duration::from_millis(33); + second + .write_i420(&frame, 320, 240, frame_time) + .await + .expect("first frame"); + let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + // VP9 payload descriptor: B starts a frame, P marks an inter frame. + let starts_a_keyframe = |payload: &[u8]| payload[0] & 0x08 != 0 && payload[0] & 0x40 == 0; + + receiver + .write_rtcp(&[Box::new(PictureLossIndication { + sender_ssrc: 0, + media_ssrc: remote.ssrc(), + })]) + .await + .expect("send PLI"); + let mut keyframes = 0; + // Ten frames are far below the periodic keyframe interval. + for _ in 0..10 { + second + .write_i420(&frame, 320, 240, frame_time) + .await + .expect("frame"); + while let Ok(Ok((packet, _))) = + tokio::time::timeout(Duration::from_millis(100), remote.read_rtp()).await + { + keyframes += usize::from(starts_a_keyframe(&packet.payload)); + } + } + + assert!( + keyframes >= 2, + "the PLI forces a keyframe after the first one (keyframes: {keyframes})" + ); + track.stop(); + reader.abort(); for task in rtcp_tasks { task.abort(); } diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index b74b91c..478f243 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -35,7 +35,7 @@ use std::sync::{Arc, Mutex as StdMutex, Weak}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use bytes::Bytes; -use tokio::sync::Semaphore; +use tokio::sync::{Notify, Semaphore}; use tokio::task::JoinHandle; use webrtc::api::media_engine::{MIME_TYPE_H264, MIME_TYPE_OPUS, MIME_TYPE_VP8, MIME_TYPE_VP9}; use webrtc::rtp::extension::HeaderExtension; @@ -118,6 +118,7 @@ struct TrackCore { fwd_seq: AtomicU16, fwd_init: AtomicBool, stopped: AtomicBool, + stop_notify: Notify, muted: AtomicBool, quality_paused: AtomicBool, track_id: String, @@ -171,6 +172,7 @@ impl TrackCore { fwd_seq: AtomicU16::new(0), fwd_init: AtomicBool::new(false), stopped: AtomicBool::new(false), + stop_notify: Notify::new(), muted: AtomicBool::new(false), quality_paused: AtomicBool::new(false), track_id, @@ -294,6 +296,18 @@ impl TrackCore { fn stop(&self) { self.stopped.store(true, Ordering::SeqCst); + self.stop_notify.notify_waiters(); + } + + /// Wait until [`Self::stop`]. + async fn stopped(&self) { + let notified = self.stop_notify.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + if self.stopped.load(Ordering::SeqCst) { + return; + } + notified.await; } fn set_muted(&self, muted: bool) { @@ -1230,6 +1244,44 @@ impl LocalVideoTrack { } } + /// Continue the RTP sequence numbers and timestamps of the single-encoding + /// `previous`, whose sender this track takes over. The encoder state moves + /// here, so the sequence numbers and the VP9 picture ids continue. The SFU + /// drops a stream whose timestamps go back. + fn continue_rtp_from(&self, previous: &LocalVideoTrack) { + let (next_pts, rtp_ts) = { + let clock = previous + .inner + .clock + .lock() + .unwrap_or_else(|e| e.into_inner()); + (clock.next_pts, clock.rtp_ts) + }; + { + let mut clock = self.inner.clock.lock().unwrap_or_else(|e| e.into_inner()); + clock.next_pts = next_pts; + clock.rtp_ts = rtp_ts; + } + let (current, previous) = (&self.inner.encodings[0], &previous.inner.encodings[0]); + let state = previous + .encoder + .lock() + .unwrap_or_else(|e| e.into_inner()) + .take(); + *current.encoder.lock().unwrap_or_else(|e| e.into_inner()) = state; + let packetizer = previous + .core + .packetizer + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + *current + .core + .packetizer + .lock() + .unwrap_or_else(|e| e.into_inner()) = packetizer; + } + /// The underlying webrtc-rs track, for attaching this track to a /// PeerConnection you manage yourself (`pc.add_track(...)`). See /// [`LocalAudioTrack::webrtc_track`]. @@ -2079,15 +2131,36 @@ impl LocalTrack { } } + /// Whether the track sends several RID encodings. + pub(crate) fn is_simulcast(&self) -> bool { + matches!(self, LocalTrack::Video { track, .. } if track.inner.encodings.len() > 1) + } + + /// Wait until the track is stopped. + pub(crate) async fn stopped(&self) { + match self { + LocalTrack::Audio(a) | LocalTrack::ScreenShareAudio(a) => a.inner.core.stopped().await, + LocalTrack::Video { track, .. } => track.inner.encodings[0].core.stopped().await, + } + } + /// Continue the RTP timeline of the audio track `previous`, whose sender /// this audio track takes over. pub(crate) fn continue_rtp_from(&self, previous: &LocalTrack) { - if let ( - LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track), - LocalTrack::Audio(previous) | LocalTrack::ScreenShareAudio(previous), - ) = (self, previous) - { - track.continue_rtp_from(previous); + match (self, previous) { + ( + LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track), + LocalTrack::Audio(previous) | LocalTrack::ScreenShareAudio(previous), + ) => track.continue_rtp_from(previous), + ( + LocalTrack::Video { track, .. }, + LocalTrack::Video { + track: previous, .. + }, + ) => { + track.continue_rtp_from(previous); + } + _ => {} } } diff --git a/src/video/call.rs b/src/video/call.rs index f6129be..6fcc9b4 100644 --- a/src/video/call.rs +++ b/src/video/call.rs @@ -997,7 +997,11 @@ impl Call { /// Stop publishing a previously published track. The publisher keeps its /// transceiver in the negotiated envelope and signals the stop to the SFU /// via `UpdateMuteStates` (no publisher renegotiation), matching - /// `stream-video-js`. + /// `stream-video-js`. A new track of the same kind that is published later + /// takes over that transceiver. A simulcast track cannot do this + /// ([`RtcError::SimulcastReplace`](crate::rtc::RtcError::SimulcastReplace)); + /// turn it off and on with [`mute_track`](Self::mute_track) and + /// [`unmute_track`](Self::unmute_track). pub async fn stop_publish(&self, track: crate::rtc::LocalTrack) -> crate::rtc::RtcResult<()> { self.rtc.stop_publish(track).await } diff --git a/tests/rtc_media.rs b/tests/rtc_media.rs index 875950d..e4f35f1 100644 --- a/tests/rtc_media.rs +++ b/tests/rtc_media.rs @@ -1483,3 +1483,129 @@ async fn sole_audio_can_be_stopped_and_published_again() { let _ = admin.delete(DeleteCallRequest { hard: Some(true) }).await; outcome.expect("test 9 (stop sole publication) timed out"); } + +/// A stopped camera track is replaced by a new track +/// +/// A publishes VP9 video, stops it, and publishes a new VP9 track on the same +/// session. The new track takes over the sender that the stop kept, as JS +/// `replaceTrack` does, so B decodes the new track on the remote track it +/// already has, and A does not reconnect. +#[tokio::test] +async fn sole_video_can_be_stopped_and_published_again() { + 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 mut events_a = call_a.client_events(); + 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.clone()); + 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"); + tokio::time::timeout(Duration::from_secs(45), remote_a.next_video_frame()) + .await + .expect("B decoded no frame before the stop") + .expect("B's video track ended before the stop"); + + call_a + .stop_publish(LocalTrack::Video { + track: video_a, + track_type: TrackType::Video, + }) + .await + .expect("A stop_publish"); + feeder.abort(); + let video_unpublished = await_track_event( + &mut events_b, + &user_a, + TrackType::Video, + false, + Duration::from_secs(45), + ) + .await; + assert!( + video_unpublished, + "B never received A's video TrackUnpublished after stop_publish" + ); + // Frames of the first track that B still has queued. + while let Ok(Some(_)) = + tokio::time::timeout(Duration::from_secs(2), remote_a.next_video_frame()).await + { + } + + let video_again = LocalVideoTrack::vp9().expect("vp9 track"); + call_a + .publish_video(video_again.clone()) + .await + .expect("A publish_video again on the same session"); + let feeder_again = spawn_blue_video(video_again); + let video_republished = await_track_event( + &mut events_b, + &user_a, + TrackType::Video, + true, + Duration::from_secs(45), + ) + .await; + assert!( + video_republished, + "B never received A's video TrackPublished after the second publish" + ); + let frame = tokio::time::timeout(Duration::from_secs(45), remote_a.next_video_frame()) + .await + .expect("B decoded no frame of the new track on its remote track") + .expect("B's video track ended after the second publish"); + feeder_again.abort(); + assert_packed_blue_frame(&frame); + let mut reconnected = false; + while let Ok(event) = events_a.try_recv() { + reconnected |= matches!( + event, + ClientCallEvent::CallingStateChanged( + CallingState::Reconnecting | CallingState::Migrating + ) + ); + } + assert!(!reconnected, "A reconnected after the second publish"); + + 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("stop and publish video again timed out"); +} From 2ec786f80adc9031c2e27608bd439dc997a5ccfa Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 5 Oct 2026 13:34:03 +0200 Subject: [PATCH 4/6] perf: skip the Opus encode while an audio track is muted --- src/rtc/tracks/local.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index 478f243..1454060 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -772,6 +772,10 @@ async fn pace_audio(track: Weak) { *slot = buf.pop_front().unwrap_or(0); } } + // A muted track sends nothing, so its audio is dropped without an encode. + if inner.core.is_output_paused() { + continue; + } // Measure what we are about to encode, including any silence fill, so a // starved pacer reports silence rather than the last spoken level. inner From e417a72212ef1931395b0352b4bf39ea17611e32 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 5 Oct 2026 14:09:52 +0200 Subject: [PATCH 5/6] fix: keep the audio RTP clock running while a paced track is muted --- src/rtc/tracks/local.rs | 49 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index 1454060..2a93860 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -773,7 +773,15 @@ async fn pace_audio(track: Weak) { } } // A muted track sends nothing, so its audio is dropped without an encode. + // The RTP clock keeps time, so the first packet after the mute shows the + // gap (RFC 3550 §5.1). if inner.core.is_output_paused() { + inner + .core + .packetizer + .lock() + .unwrap_or_else(|e| e.into_inner()) + .skip_samples(FRAME_SAMPLES_20MS as u32); continue; } // Measure what we are about to encode, including any silence fill, so a @@ -2491,6 +2499,47 @@ mod tests { let _ = receiver.close().await; } + #[tokio::test] + async fn the_rtp_clock_runs_while_a_paced_track_is_muted() { + let track = LocalAudioTrack::opus().expect("opus track"); + let sender = peer::new_peer_connection(&[]).await.expect("sender"); + sender + .add_track(track.webrtc_track()) + .await + .expect("add track"); + let (receiver, mut remote_rx) = peer::connect_audio_receiver(&sender).await; + track.start_pacing().await; + let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + let published = LocalTrack::Audio(track.clone()); + + published.set_muted(true); + let (mut before, _) = remote.read_rtp().await.expect("packet before the mute"); + while let Ok(Ok((packet, _))) = + tokio::time::timeout(Duration::from_millis(100), remote.read_rtp()).await + { + before = packet; + } + tokio::time::sleep(Duration::from_millis(500)).await; + published.set_muted(false); + let (after, _) = tokio::time::timeout(Duration::from_secs(5), remote.read_rtp()) + .await + .expect("packet after the mute") + .expect("packet after the mute"); + + // The mute lasts at least 600 ms; a slow runtime can skip some ticks. + let skipped = after.header.timestamp.wrapping_sub(before.header.timestamp); + assert!( + skipped >= 10 * FRAME_SAMPLES_20MS as u32, + "the timestamp advanced {skipped} samples over the mute" + ); + track.stop(); + let _ = sender.close().await; + let _ = receiver.close().await; + } + #[tokio::test] async fn pcm_written_before_pacing_starts_stays_queued() { let track = LocalAudioTrack::opus_with_config( From 3fa18da21158e94fec485016e7b33c9a3b6cc15e Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 5 Oct 2026 14:19:36 +0200 Subject: [PATCH 6/6] fix: keep the RTP clock running while encoded audio or video is muted --- src/rtc/tracks/local.rs | 149 ++++++++++++++++++++++++++++++++++++---- 1 file changed, 135 insertions(+), 14 deletions(-) diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index 2a93860..6605ee1 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -213,6 +213,11 @@ impl TrackCore { )); } if self.muted.load(Ordering::SeqCst) || self.quality_paused.load(Ordering::SeqCst) { + // The RTP clock keeps time while no packet goes out. + self.packetizer + .lock() + .unwrap_or_else(|e| e.into_inner()) + .skip_samples(samples); return Ok(()); } let payload = Bytes::copy_from_slice(payload); @@ -872,6 +877,17 @@ struct VideoClock { rtp_ts: u32, } +impl VideoClock { + /// Return the presentation time and RTP timestamp of the next frame, and + /// move the clock past its duration. + fn advance(&mut self, dur_ms: i64, samples: u32) -> (i64, u32) { + let current = (self.next_pts, self.rtp_ts); + self.next_pts = self.next_pts.saturating_add(dur_ms); + self.rtp_ts = self.rtp_ts.wrapping_add(samples); + current + } +} + struct VideoEncoding { core: TrackCore, encoder: StdMutex>, @@ -1138,6 +1154,9 @@ impl LocalVideoTrack { "write to a stopped track".to_owned(), )); } + let dur_ms = i64::try_from(duration.as_millis().max(1)).unwrap_or(i64::MAX); + let samples = + (duration.as_secs_f64() * f64::from(self.inner.encodings[0].core.clock_rate)) as u32; if self .inner .encodings @@ -1147,6 +1166,12 @@ impl LocalVideoTrack { )) .all(|encoding| encoding.core.is_output_paused()) { + // The RTP clock keeps time while no frame goes out. + self.inner + .clock + .lock() + .unwrap_or_else(|e| e.into_inner()) + .advance(dur_ms, samples); return Ok(()); } if width == 0 || height == 0 || !width.is_multiple_of(2) || !height.is_multiple_of(2) { @@ -1205,10 +1230,6 @@ impl LocalVideoTrack { ))); } - let dur_ms = i64::try_from(duration.as_millis().max(1)).unwrap_or(i64::MAX); - let samples = - (duration.as_secs_f64() * f64::from(self.inner.encodings[0].core.clock_rate)) as u32; - let permit = self .inner .encode_gate @@ -1666,16 +1687,11 @@ fn encode_i420_layers( dur_ms: i64, samples: u32, ) -> Result)>> { - let (pts, timestamp) = { - let mut clock = inner - .clock - .lock() - .unwrap_or_else(|error| error.into_inner()); - let current = (clock.next_pts, clock.rtp_ts); - clock.next_pts = clock.next_pts.saturating_add(dur_ms); - clock.rtp_ts = clock.rtp_ts.wrapping_add(samples); - current - }; + let (pts, timestamp) = inner + .clock + .lock() + .unwrap_or_else(|error| error.into_inner()) + .advance(dur_ms, samples); let count = usize::from(inner.active_encoding_count.load(Ordering::SeqCst)); let mut output = Vec::with_capacity(count); for (index, encoding) in inner.encodings.iter().take(count).enumerate() { @@ -2540,6 +2556,111 @@ mod tests { let _ = receiver.close().await; } + #[tokio::test] + async fn the_rtp_clock_runs_while_encoded_audio_is_muted() { + let track = LocalAudioTrack::opus().expect("opus track"); + let sender = peer::new_peer_connection(&[]).await.expect("sender"); + sender + .add_track(track.webrtc_track()) + .await + .expect("add track"); + let (receiver, mut remote_rx) = peer::connect_audio_receiver(&sender).await; + let silence = [0xf8u8, 0xff, 0xfe]; + let frame = Duration::from_millis(20); + let published = LocalTrack::Audio(track.clone()); + + for _ in 0..3 { + track.write_sample(&silence, frame).await.expect("write"); + } + let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + let mut before = None; + while let Ok(Ok((packet, _))) = + tokio::time::timeout(Duration::from_millis(100), remote.read_rtp()).await + { + before = Some(packet); + } + published.set_muted(true); + for _ in 0..10 { + track + .write_sample(&silence, frame) + .await + .expect("muted write"); + } + published.set_muted(false); + track.write_sample(&silence, frame).await.expect("write"); + let (after, _) = tokio::time::timeout(Duration::from_secs(5), remote.read_rtp()) + .await + .expect("packet after the mute") + .expect("packet after the mute"); + + let before = before.expect("packets before the mute"); + assert_eq!( + after.header.timestamp.wrapping_sub(before.header.timestamp), + 11 * FRAME_SAMPLES_20MS as u32 + ); + let _ = sender.close().await; + let _ = receiver.close().await; + } + + #[tokio::test] + async fn the_rtp_clock_runs_while_a_video_track_is_muted() { + use webrtc::rtp_transceiver::rtp_codec::RTPCodecType; + + let track = LocalVideoTrack::vp9().expect("vp9 track"); + let sender = peer::new_peer_connection(&[]).await.expect("sender"); + sender + .add_track(track.webrtc_track()) + .await + .expect("add track"); + let (receiver, mut remote_rx) = peer::connect_receiver(&sender, RTPCodecType::Video).await; + let frame = vec![128_u8; 320 * 240 * 3 / 2]; + let frame_time = Duration::from_millis(33); + + for _ in 0..3 { + track + .write_i420(&frame, 320, 240, frame_time) + .await + .expect("frame"); + } + let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + let mut before = None; + while let Ok(Ok((packet, _))) = + tokio::time::timeout(Duration::from_millis(100), remote.read_rtp()).await + { + before = Some(packet); + } + track.set_muted(true); + for _ in 0..10 { + track + .write_i420(&frame, 320, 240, frame_time) + .await + .expect("muted frame"); + } + track.set_muted(false); + track + .write_i420(&frame, 320, 240, frame_time) + .await + .expect("frame"); + let (after, _) = tokio::time::timeout(Duration::from_secs(5), remote.read_rtp()) + .await + .expect("packet after the mute") + .expect("packet after the mute"); + + let before = before.expect("packets before the mute"); + assert_eq!( + after.header.timestamp.wrapping_sub(before.header.timestamp), + 11 * 2_970 + ); + let _ = sender.close().await; + let _ = receiver.close().await; + } + #[tokio::test] async fn pcm_written_before_pacing_starts_stays_queued() { let track = LocalAudioTrack::opus_with_config(