feat/pcm pacing control

CHANGELOG.md

@@ -74,8 +74,34 @@ 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 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

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())
@@ -334,8 +336,6 @@ async fn configure_openai(
         Box::pin(async move {
             if state == RTCPeerConnectionState::Connected {
                 paced_mic.start_pacing().await;
-            } else {
-                paced_mic.pause_pacing();
             }
         })
     }));

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
     }

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(

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 {

src/rtc/join/publication.rs

@@ -27,7 +27,7 @@ pub(super) struct LocalPublication {
 pub(super) struct MediaState {
     pub(super) publications: Vec<LocalPublication>,
     pub(super) publish_quality: HashMap<(i32, i32), Vec<event::VideoLayerSetting>>,
-    /// 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)>,
 }
@@ -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<LocalTrack> {
         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,
@@ -96,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<LocalTrack> {
         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)
     }
 
@@ -118,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<LocalTrack> {
-        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<usize> {
+        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],

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,
@@ -166,21 +169,23 @@ impl RtcCore {
         Ok(())
     }
 
-    /// Pace the published audio only while the current publisher is connected.
-    pub(super) async fn sync_audio_pacing(&self) {
+    /// 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
             .publisher_handles()
             .await
             .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;
         }
     }
 
@@ -206,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<Self>, track: LocalTrack) -> Result<()> {
         let mut media = self.media.lock().await;
         let track_id = track.track_id();

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