From a8d7cd6f7675163e8fd3856d3969b11c0610e9a9 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 12 Oct 2026 14:16:34 +0200 Subject: [PATCH 1/7] fix: let flush drop a write in progress without a wait flush took the resampler lock with a blocking lock, and a write holds that lock for the full resample: about 130 ms for a 30 s write at 24 kHz. Also, a flush that came before a write took the lock did not remove the audio of that write. A flush counter now marks the writes that started before a flush: their audio is not queued, and the write resets the filter. flush no longer waits for the resampler. --- src/rtc/tracks/local.rs | 103 ++++++++++++++++++++++++++++++++++------ 1 file changed, 89 insertions(+), 14 deletions(-) diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index 80a7ec5..9ee2eb7 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -34,7 +34,9 @@ use std::collections::VecDeque; use std::num::NonZeroU8; -use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU8, AtomicU16, AtomicU32, Ordering}; +use std::sync::atomic::{ + AtomicBool, AtomicI32, AtomicU8, AtomicU16, AtomicU32, AtomicU64, Ordering, +}; use std::sync::{Arc, Mutex as StdMutex, OnceLock, TryLockError, Weak}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -354,9 +356,12 @@ struct AudioInner { /// Resampled 48 kHz mono PCM awaiting the 20 ms pacer. pcm: StdMutex>, pcm_capacity_samples: usize, - /// Locked before `pcm`. A writer keeps it until its audio is queued, so a - /// flush cannot fall between the resample and the queue. + /// Locked before `pcm`. A writer keeps it until its audio is queued. + /// `flush` does not wait for it. resampler: StdMutex, + /// Counts `flush` calls. Changed and compared under the `pcm` lock. A + /// write queues no audio when a flush came after the write started. + flushes: AtomicU64, /// Set by `write_pcm` and cleared at each pacer tick. A resampler flush /// makes the next write start with silence. pcm_written: AtomicBool, @@ -557,6 +562,7 @@ impl LocalAudioTrack { pcm: StdMutex::new(VecDeque::new()), pcm_capacity_samples, resampler: StdMutex::new(StreamResampler::new(OPUS_SAMPLE_RATE, 1)), + flushes: AtomicU64::new(0), pcm_written: AtomicBool::new(false), encoder: StdMutex::new(encoder), pacer: StdMutex::new(None), @@ -601,6 +607,7 @@ impl LocalAudioTrack { "write to a stopped track".to_owned(), )); } + let flushes = self.inner.flushes.load(Ordering::SeqCst); self.inner.pcm_pacing.store(true, Ordering::SeqCst); let frame = if frame.channels > 1 { let channels = usize::from(frame.channels); @@ -626,7 +633,12 @@ impl LocalAudioTrack { .unwrap_or_else(|e| e.into_inner()); let resampled = r.push(&frame)?.samples; track.inner.pcm_written.store(true, Ordering::SeqCst); - Ok(track.queue_pcm(resampled)) + let queued = track.queue_pcm(resampled, flushes); + // After a flush, the filter must not keep the audio of this write. + if track.inner.flushes.load(Ordering::SeqCst) != flushes { + r.flush()?; + } + Ok(queued) }) .await .map_err(|error| RtcError::Media(format!("pcm resample worker failed: {error}")))??; @@ -640,9 +652,13 @@ impl LocalAudioTrack { } /// Append 48 kHz mono samples to the pacer queue, dropping the oldest - /// audio above the capacity. Callers hold the `resampler` lock. - fn queue_pcm(&self, samples: Vec) -> Result<()> { + /// audio above the capacity. Queues nothing when a flush came after + /// `flushes` was read. Callers hold the `resampler` lock. + fn queue_pcm(&self, samples: Vec, flushes: u64) -> Result<()> { let mut buf = self.inner.pcm.lock().unwrap_or_else(|e| e.into_inner()); + if self.inner.flushes.load(Ordering::SeqCst) != flushes { + return Ok(()); + } let dropped = push_bounded_pcm(&mut buf, samples, self.inner.pcm_capacity_samples); if dropped == 0 { return Ok(()); @@ -722,18 +738,23 @@ impl LocalAudioTrack { } /// Drop any buffered-but-unsent PCM (barge-in), including the audio that - /// the resampler still holds. Does not stop the track. + /// the resampler still holds and the audio of a + /// [`write_pcm`](Self::write_pcm) that started before this call. Does not + /// wait for a write in progress. Does not stop the track. pub fn flush(&self) { - let mut r = self - .inner - .resampler - .lock() - .unwrap_or_else(|e| e.into_inner()); + let mut buf = self.inner.pcm.lock().unwrap_or_else(|e| e.into_inner()); + self.inner.flushes.fetch_add(1, Ordering::SeqCst); + buf.clear(); + drop(buf); + // A write that holds the resampler resets it itself. + let mut r = match self.inner.resampler.try_lock() { + Ok(r) => r, + Err(TryLockError::Poisoned(e)) => e.into_inner(), + Err(TryLockError::WouldBlock) => return, + }; if let Err(e) = r.flush() { tracing::debug!(error = %e, "stream.rtc.audio.flush_failed"); } - let mut buf = self.inner.pcm.lock().unwrap_or_else(|e| e.into_inner()); - buf.clear(); } /// Start taking queued PCM: one 20 ms frame every 20 ms, or silence when @@ -2271,6 +2292,8 @@ impl LocalTrack { #[cfg(test)] mod tests { + use futures_util::FutureExt; + use super::*; use crate::rtc::peer; use crate::rtc::proto::models::{Codec, VideoDimension}; @@ -2570,6 +2593,58 @@ mod tests { ); } + #[test] + fn flush_returns_while_a_write_resamples() { + let track = unpaced_track(); + let writing = track + .inner + .resampler + .lock() + .unwrap_or_else(|e| e.into_inner()); + let (flushed_tx, flushed_rx) = std::sync::mpsc::channel(); + let flushing = track.clone(); + std::thread::spawn(move || { + flushing.flush(); + let _ = flushed_tx.send(()); + }); + + let flushed = flushed_rx.recv_timeout(Duration::from_secs(5)); + drop(writing); + + flushed.expect("the flush waited for the write"); + } + + #[test] + fn flush_drops_a_write_that_waits_for_a_blocking_thread() { + let runtime = tokio::runtime::Builder::new_current_thread() + .max_blocking_threads(1) + .build() + .expect("runtime"); + runtime.block_on(async { + let track = unpaced_track(); + let (release, released) = std::sync::mpsc::channel::<()>(); + // Takes the only blocking thread, so the write cannot resample yet. + let _busy = tokio::task::spawn_blocking(move || released.recv()); + let mut write = std::pin::pin!(track.write_pcm(loud_end_16k())); + assert!(write.as_mut().now_or_never().is_none()); + + track.flush(); + drop(release); + write.await.expect("write_pcm"); + let mut first = vec![0; FRAME_SAMPLES_20MS]; + let mut second = vec![0; FRAME_SAMPLES_20MS]; + track.inner.take_pcm(&mut first); + track.inner.take_pcm(&mut second); + + assert_eq!( + peak(&first).max(peak(&second)), + 0, + "audio from before the flush played" + ); + track.stop(); + }); + } + #[test] fn pcm_queue_overflow_drops_oldest_and_flush_still_clears() { let track = LocalAudioTrack::opus().expect("opus track"); From f2491778d9de24880831c3179b4079c57c384578 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 12 Oct 2026 14:16:34 +0200 Subject: [PATCH 2/7] docs: describe the flush of a write in progress in the changelog --- CHANGELOG.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 31238bb..9b8768d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -230,8 +230,9 @@ on `Call` (`get_call_participant_session_metrics`, 48 kHz to 16 kHz, a 12 kHz tone came out at 4 kHz at full level. `write_pcm` runs the filter on a blocking thread, so a long frame does not stop the runtime or the pacer. -- `LocalAudioTrack::flush` also drops the audio in the resampler. A flush during - a `write_pcm` can no longer put audio from before the flush in the queue. +- `LocalAudioTrack::flush` also drops the audio in the resampler and the audio + of a `write_pcm` that started before the flush. It does not wait for a + `write_pcm` in progress. # v0.1.0-preview.2 From 589c34b112966e40b64dfbbd524a29303d40539c Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 12 Oct 2026 15:33:25 +0200 Subject: [PATCH 3/7] fix: share one RTP timeline across the write paths of a track Each write path kept its own counters: the packetizer for write_pcm and write_sample, VideoEncoder.seq and VideoClock.rtp_ts for write_i420, and fwd_seq with the source timestamp for write_rtp. continue_rtp_from did not carry the forward state over. When a forwarding track took over a stopped sender, or a track took over the sender of a forwarding track, the sequence numbers jumped or went back on the same SSRC, and the receiver dropped the packets. TrackCore now holds one RtpTimeline (last sequence number, next timestamp, forward offset) for all write paths. Packetized writes take their sequence numbers and timestamps from it. Forwarded packets keep the spacing of the source timestamps, shifted to continue the timeline. continue_rtp_from carries the timeline over and adds the time since the stop. --- src/rtc/peer/publisher.rs | 254 ++++++++++++++++++++++++++++++++++- src/rtc/tracks/local.rs | 272 ++++++++++++++++++++++---------------- 2 files changed, 410 insertions(+), 116 deletions(-) diff --git a/src/rtc/peer/publisher.rs b/src/rtc/peer/publisher.rs index 23a8818..666f9c3 100644 --- a/src/rtc/peer/publisher.rs +++ b/src/rtc/peer/publisher.rs @@ -370,8 +370,10 @@ mod tests { use crate::rtc::peer; use crate::rtc::proto::event::VideoLayerSetting; use crate::rtc::proto::models::{Codec, VideoDimension}; - use crate::rtc::tracks::{LocalAudioTrack, LocalVideoTrack}; + use crate::rtc::tracks::{LocalAudioTrack, LocalVideoTrack, RtpPacket}; + use bytes::Bytes; use std::time::{Duration, Instant}; + use webrtc::rtp::header::Header; use webrtc::rtp_transceiver::rtp_codec::RTPCodecType; #[tokio::test] @@ -437,6 +439,171 @@ mod tests { let _ = receiver.close().await; } + #[tokio::test] + async fn a_forwarding_track_continues_the_rtp_timeline_of_a_retired_pcm_track() { + let takeover = take_over_audio_sender(AudioWriter::Pcm, AudioWriter::Rtp).await; + + // One 20 ms frame, then the time from the stop to the replace. + takeover.assert_continues_after(960); + } + + #[tokio::test] + async fn a_forwarding_track_continues_the_rtp_timeline_of_a_retired_forwarding_track() { + let takeover = take_over_audio_sender(AudioWriter::Rtp, AudioWriter::Rtp).await; + + // The time from the stop to the replace. + takeover.assert_continues_after(0); + } + + #[tokio::test] + async fn a_pcm_track_continues_the_rtp_timeline_of_a_retired_forwarding_track() { + let takeover = take_over_audio_sender(AudioWriter::Rtp, AudioWriter::Pcm).await; + + // The time from the stop to the replace. + takeover.assert_continues_after(0); + } + + /// How a test track writes its audio. + #[derive(Clone, Copy)] + enum AudioWriter { + /// The pacer of `write_pcm`. + Pcm, + /// `write_rtp` of a source that has its own RTP timeline. + Rtp, + } + + impl AudioWriter { + /// Write one 20 ms frame every 20 ms until `track` stops. A source + /// starts at the RTP timestamp `source_timestamp`. + async fn start(self, track: &LocalAudioTrack, source_timestamp: u32) { + match self { + AudioWriter::Pcm => track.start_pacing().await, + AudioWriter::Rtp => { + let track = track.clone(); + let mut packet = RtpPacket { + header: Header { + version: 2, + payload_type: 111, + timestamp: source_timestamp, + ..Default::default() + }, + payload: Bytes::from_static(&[0xf8, 0xff, 0xfe]), + }; + tokio::spawn(async move { + while track.write_rtp(packet.clone()).await.is_ok() { + packet.header.sequence_number = + packet.header.sequence_number.wrapping_add(1); + packet.header.timestamp = packet.header.timestamp.wrapping_add(960); + tokio::time::sleep(Duration::from_millis(20)).await; + } + }); + } + } + } + } + + /// The last packet of a stopped audio track and the first packet of the + /// track that takes over its sender. + struct AudioTakeover { + last: RtpPacket, + next: RtpPacket, + /// The minimum time from the stop to the replace. + waited: Duration, + /// The maximum time from the stop to the replace. + replaced: Duration, + } + + impl AudioTakeover { + /// Assert that `next` follows `last`, and that its timestamp advanced + /// by `frame` samples plus the time from the stop to the replace. + fn assert_continues_after(&self, frame: u32) { + assert_eq!( + self.next.header.sequence_number, + self.last.header.sequence_number.wrapping_add(1) + ); + let expected = |gap: Duration| frame + (gap.as_secs_f64() * 48_000.0) as u32; + let advanced = self + .next + .header + .timestamp + .wrapping_sub(self.last.header.timestamp); + assert!( + (expected(self.waited)..=expected(self.replaced)).contains(&advanced), + "the timestamp advanced {advanced} samples over a stop of {:?} to {:?}", + self.waited, + self.replaced + ); + } + } + + /// Stop an audio track that `first` writes, put a track that `second` + /// writes on its sender, and read the packets around the replace. + async fn take_over_audio_sender(first: AudioWriter, second: AudioWriter) -> AudioTakeover { + let opus = [PublishOption { + track_type: TrackType::Audio as i32, + codec: Some(Codec { + name: "opus".to_owned(), + ..Default::default() + }), + ..Default::default() + }]; + let first_track = LocalAudioTrack::opus().expect("first track"); + let retired = LocalTrack::Audio(first_track.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; + first.start(&first_track, 0).await; + let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + let (mut last, _) = remote.read_rtp().await.expect("first packet"); + let before_stop = Instant::now(); + retired.stop(); + let stopped = Instant::now(); + while let Ok(Ok((packet, _))) = + tokio::time::timeout(Duration::from_millis(200), remote.read_rtp()).await + { + last = packet; + } + let waited = stopped.elapsed(); + + let second_track = LocalAudioTrack::opus().expect("second track"); + let track = LocalTrack::Audio(second_track.clone()); + let reader = replace_retired_track(&publisher, &retired, &track, &opus) + .await + .expect("replace the retired track") + .expect("the retired sender"); + let replaced = before_stop.elapsed(); + // A source timeline that is far from the timeline of the first track. + second + .start( + &second_track, + last.header.timestamp.wrapping_add(u32::MAX / 2), + ) + .await; + 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"); + + track.stop(); + reader.abort(); + for task in rtcp_tasks { + task.abort(); + } + let _ = publisher.close().await; + let _ = receiver.close().await; + AudioTakeover { + last, + next, + waited, + replaced, + } + } + #[tokio::test] async fn a_video_track_on_a_retired_sender_continues_its_rtp_timeline() { let vp9 = [video_option("VP9")]; @@ -524,6 +691,91 @@ mod tests { let _ = receiver.close().await; } + #[tokio::test] + async fn an_i420_track_continues_the_rtp_timeline_of_a_retired_forwarding_track() { + 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 forwarding = first.clone(); + tokio::spawn(async move { + let mut packet = RtpPacket { + header: Header { + version: 2, + marker: true, + ..Default::default() + }, + payload: Bytes::from_static(&[0x08, 0x00]), + }; + while forwarding.write_rtp(packet.clone()).await.is_ok() { + packet.header.sequence_number = packet.header.sequence_number.wrapping_add(1); + packet.header.timestamp = packet.header.timestamp.wrapping_add(2_970); + tokio::time::sleep(Duration::from_millis(33)).await; + } + }); + let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + let (mut last, _) = remote.read_rtp().await.expect("first packet"); + let before_stop = Instant::now(); + retired.stop(); + let stopped = Instant::now(); + while let Ok(Ok((packet, _))) = + tokio::time::timeout(Duration::from_millis(200), remote.read_rtp()).await + { + last = packet; + } + let waited = stopped.elapsed(); + + 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 replaced = before_stop.elapsed(); + let frame = vec![128_u8; 320 * 240 * 3 / 2]; + second + .write_i420(&frame, 320, 240, Duration::from_millis(33)) + .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) + ); + // The time from the stop to the replace. + let expected = |gap: Duration| (gap.as_secs_f64() * 90_000.0) as u32; + let advanced = next.header.timestamp.wrapping_sub(last.header.timestamp); + assert!( + (expected(waited)..=expected(replaced)).contains(&advanced), + "the timestamp advanced {advanced} samples over a stop of {waited:?} to {replaced:?}" + ); + 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; diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index 9ee2eb7..eda7c9b 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -21,7 +21,11 @@ //! packetizes and writes. The caller controls pacing. //! - `write_rtp` — forward an inbound [`RtpPacket`]. The SDK allocates a fresh //! SSRC (webrtc-rs binding) and **rewrites the sequence number**, dropping the -//! inbound header extensions; callers must not reuse inbound headers. +//! inbound header extensions; callers must not reuse inbound headers. The +//! timestamps keep the spacing of the inbound ones. +//! +//! The write paths of a track share one RTP timeline. A track that takes over +//! the sender of a stopped track continues the timeline of that track. //! //! Outbound audio packets carry the RFC 6464 `ssrc-audio-level` extension: the //! SFU marks a participant speaking only from that level, and remote UIs draw @@ -34,9 +38,7 @@ use std::collections::VecDeque; use std::num::NonZeroU8; -use std::sync::atomic::{ - AtomicBool, AtomicI32, AtomicU8, AtomicU16, AtomicU32, AtomicU64, Ordering, -}; +use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU8, AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Mutex as StdMutex, OnceLock, TryLockError, Weak}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -111,16 +113,15 @@ pub fn audio_level_dbov(rms: f64) -> u8 { } /// Shared outbound-track machinery: the webrtc track, its packetizer, and the -/// forward-path sequence counter. +/// RTP timeline of all write paths. struct TrackCore { track: Arc, /// Packetizer for the encoded-sample paths (`write_pcm` pacer + `write_sample`). + /// `rtp` replaces its sequence numbers and timestamps. packetizer: StdMutex>, clock_rate: u32, mime_type: String, - /// Monotonic sequence number for the RTP-forward path. - fwd_seq: AtomicU16, - fwd_init: AtomicBool, + rtp: StdMutex, stopped: AtomicBool, /// The time of the first [`Self::stop`]. stopped_at: OnceLock, @@ -175,8 +176,11 @@ impl TrackCore { packetizer: StdMutex::new(Box::new(packetizer)), clock_rate, mime_type, - fwd_seq: AtomicU16::new(0), - fwd_init: AtomicBool::new(false), + rtp: StdMutex::new(RtpTimeline { + seq: seed_u16(), + timestamp: None, + forward_offset: None, + }), stopped: AtomicBool::new(false), stopped_at: OnceLock::new(), stop_notify: Notify::new(), @@ -221,17 +225,25 @@ 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 + self.rtp .lock() .unwrap_or_else(|e| e.into_inner()) - .skip_samples(samples); + .advance(samples); return Ok(()); } let payload = Bytes::copy_from_slice(payload); let packets = { let mut p = self.packetizer.lock().unwrap_or_else(|e| e.into_inner()); - p.packetize(&payload, samples) - .map_err(|e| RtcError::Media(format!("packetize: {e}")))? + let mut packets = p + .packetize(&payload, samples) + .map_err(|e| RtcError::Media(format!("packetize: {e}")))?; + let mut rtp = self.rtp.lock().unwrap_or_else(|e| e.into_inner()); + let timestamp = rtp.advance(samples); + for pkt in &mut packets { + pkt.header.sequence_number = rtp.next_seq(); + pkt.header.timestamp = timestamp; + } + packets }; // `write_rtp_with_extensions` resolves each URI to the id negotiated for // the binding, so extension ids are never ours to manage. An empty slice @@ -266,7 +278,9 @@ impl TrackCore { } /// Forward an inbound RTP packet under a fresh SSRC (binding) with a rewritten - /// sequence number and no inbound header extensions. + /// sequence number and no inbound header extensions. The timestamps keep + /// the spacing of the inbound ones and continue the RTP timeline of the + /// track; a new track without an earlier write keeps them unchanged. /// /// Inbound extension ids are negotiated per-PeerConnection, so carrying them /// across would be meaningless; they are stripped and the sender re-adds the @@ -280,11 +294,18 @@ impl TrackCore { "write to a stopped track".to_owned(), )); } - if self.muted.load(Ordering::SeqCst) || self.quality_paused.load(Ordering::SeqCst) { - return Ok(()); - } + let (sequence_number, timestamp) = { + let mut rtp = self.rtp.lock().unwrap_or_else(|e| e.into_inner()); + // The RTP clock keeps time while no packet goes out. + let timestamp = rtp.forward(packet.header.timestamp); + if self.muted.load(Ordering::SeqCst) || self.quality_paused.load(Ordering::SeqCst) { + return Ok(()); + } + (rtp.next_seq(), timestamp) + }; let mut pkt = packet.clone(); - pkt.header.sequence_number = self.next_fwd_seq(); + pkt.header.sequence_number = sequence_number; + pkt.header.timestamp = timestamp; pkt.header.extension = false; pkt.header.extension_profile = 0; pkt.header.extensions.clear(); @@ -296,14 +317,27 @@ impl TrackCore { Ok(()) } - fn next_fwd_seq(&self) -> u16 { - if !self.fwd_init.swap(true, Ordering::SeqCst) { - let base = seed_u16(); - self.fwd_seq.store(base, Ordering::SeqCst); - base - } else { - self.fwd_seq.fetch_add(1, Ordering::SeqCst).wrapping_add(1) - } + /// Continue the RTP timeline and the payloader state of `previous`, whose + /// sender this track takes over. The timestamps also count the time since + /// `previous` stopped. + fn continue_rtp_from(&self, previous: &TrackCore) { + let stopped_for = previous + .stopped_at + .get() + .map_or(Duration::ZERO, Instant::elapsed); + let packetizer = previous + .packetizer + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + *self.packetizer.lock().unwrap_or_else(|e| e.into_inner()) = packetizer; + let samples = (stopped_for.as_secs_f64() * f64::from(self.clock_rate)) as u32; + let mut rtp = *previous.rtp.lock().unwrap_or_else(|e| e.into_inner()); + rtp.timestamp = rtp + .timestamp + .map(|timestamp| timestamp.wrapping_add(samples)); + rtp.forward_offset = None; + *self.rtp.lock().unwrap_or_else(|e| e.into_inner()) = rtp; } fn stop(&self) { @@ -340,8 +374,48 @@ impl TrackCore { } } +/// The RTP sequence numbers and timestamps of a sender. All write paths of a +/// track use it, and a track that takes over the sender continues it. +#[derive(Clone, Copy)] +struct RtpTimeline { + /// The last sequence number. + seq: u16, + /// The timestamp of the next frame, or of the last forwarded packet. + /// `None` until the first write. + timestamp: Option, + /// Added to the timestamps of forwarded packets. Set by the first one. + forward_offset: Option, +} + +impl RtpTimeline { + fn next_seq(&mut self) -> u16 { + self.seq = self.seq.wrapping_add(1); + self.seq + } + + /// Return the timestamp of a frame of `samples`, and move past it. + fn advance(&mut self, samples: u32) -> u32 { + let timestamp = self.timestamp.unwrap_or_else(|| u32::from(seed_u16())); + self.timestamp = Some(timestamp.wrapping_add(samples)); + timestamp + } + + /// Return the timestamp of a forwarded packet with the inbound timestamp + /// `source`. The first forwarded packet gets the timestamp of the next + /// frame, or keeps `source` when there was no earlier write. + fn forward(&mut self, source: u32) -> u32 { + let offset = *self + .forward_offset + .get_or_insert_with(|| self.timestamp.map_or(0, |next| next.wrapping_sub(source))); + let timestamp = source.wrapping_add(offset); + self.timestamp = Some(timestamp); + timestamp + } +} + /// A low-entropy 16-bit seed from the wall clock (no `rand` dependency), used to -/// randomize the forward-path sequence base so it does not always start at 0. +/// randomize the first RTP sequence numbers and timestamps so they do not always +/// start at 0. fn seed_u16() -> u16 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -716,7 +790,10 @@ impl LocalAudioTrack { } /// Forward an inbound RTP packet (same-codec republish). The SDK rewrites the - /// SSRC (new binding) and sequence number and drops inbound extensions. + /// SSRC (new binding) and sequence number and drops inbound extensions. The + /// timestamps keep the spacing of the inbound ones. On a sender that this + /// track took over, or after another write path, they continue the RTP + /// timeline of the earlier packets. /// /// The forwarded packets carry no audio level: the SFU does not forward one /// downstream, so there is nothing to preserve. Use @@ -792,31 +869,11 @@ impl LocalAudioTrack { } /// Continue the RTP sequence numbers and timestamps of `previous`, whose - /// sender this track takes over. The SFU drops a stream whose timestamps - /// go back. The timestamps also count the time since `previous` stopped. + /// sender this track takes over, for all write paths of the two tracks. + /// The SFU drops a stream whose timestamps go back. The timestamps also + /// count the time since `previous` stopped. fn continue_rtp_from(&self, previous: &LocalAudioTrack) { - let mut packetizer = previous - .inner - .core - .packetizer - .lock() - .unwrap_or_else(|e| e.into_inner()) - .clone(); - let stopped_for = previous - .inner - .core - .stopped_at - .get() - .map_or(Duration::ZERO, Instant::elapsed); - packetizer.skip_samples( - (stopped_for.as_secs_f64() * f64::from(self.inner.core.clock_rate)) as u32, - ); - *self - .inner - .core - .packetizer - .lock() - .unwrap_or_else(|e| e.into_inner()) = packetizer; + self.inner.core.continue_rtp_from(&previous.inner.core); } /// Stop the pacer and reject further writes. Called by `stop_publish`/`leave`. @@ -907,10 +964,10 @@ async fn pace_audio(track: Weak) { if !taking || inner.core.is_output_paused() { inner .core - .packetizer + .rtp .lock() .unwrap_or_else(|e| e.into_inner()) - .skip_samples(FRAME_SAMPLES_20MS as u32); + .advance(FRAME_SAMPLES_20MS as u32); continue; } // Measure what we are about to encode, including any silence fill, so a @@ -970,8 +1027,8 @@ enum VideoCodecState { }, } -/// A video encoder plus its presentation clock, RTP packetizer, and outbound -/// RTP counters. Native codec access is serialized by the enclosing mutex. +/// A video encoder plus its presentation clock and RTP packetizer. Native +/// codec access is serialized by the enclosing mutex. struct VideoEncoder { codec: VideoCodecState, width: u32, @@ -981,22 +1038,20 @@ struct VideoEncoder { last_key_pts: i64, /// Presentation time of the last frame emitted for max-framerate control. last_emit_pts: Option, - /// Outbound RTP sequence number (wraps). - seq: u16, } +/// The presentation clock. The RTP timestamps of all encodings come from the +/// RTP timeline of the first encoding. struct VideoClock { next_pts: i64, - 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); + /// Return the presentation time of the next frame, and move the clock past + /// its duration. + fn advance(&mut self, dur_ms: i64) -> i64 { + let current = self.next_pts; self.next_pts = self.next_pts.saturating_add(dur_ms); - self.rtp_ts = self.rtp_ts.wrapping_add(samples); current } } @@ -1202,10 +1257,7 @@ impl LocalVideoTrack { encodings, codec_id, encode_gate: Arc::new(Semaphore::new(1)), - clock: StdMutex::new(VideoClock { - next_pts: 0, - rtp_ts: u32::from(seed_u16()), - }), + clock: StdMutex::new(VideoClock { next_pts: 0 }), active_encoding_count: AtomicU8::new(1), layering: config.layering, planned_width: AtomicU32::new(0), @@ -1255,7 +1307,13 @@ impl LocalVideoTrack { .clock .lock() .unwrap_or_else(|e| e.into_inner()) - .advance(dur_ms, samples); + .advance(dur_ms); + self.inner.encodings[0] + .core + .rtp + .lock() + .unwrap_or_else(|e| e.into_inner()) + .advance(samples); return Ok(()); } if width == 0 || height == 0 || !width.is_multiple_of(2) || !height.is_multiple_of(2) { @@ -1346,7 +1404,11 @@ impl LocalVideoTrack { .await } - /// Forward an inbound RTP packet (same-codec republish). + /// Forward an inbound RTP packet (same-codec republish). The SDK rewrites the + /// SSRC and sequence number and drops inbound extensions. The timestamps + /// keep the spacing of the inbound ones. On a sender that this track took + /// over, or after another write path, they continue the RTP timeline of + /// the earlier packets. pub async fn write_rtp(&self, packet: RtpPacket) -> Result<()> { self.ensure_single_layer_input("RTP forwarding")?; self.inner.encodings[0].core.forward_rtp(&packet).await @@ -1359,34 +1421,26 @@ 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. The timestamps also count the - /// time since `previous` stopped. + /// `previous`, whose sender this track takes over, for all write paths of + /// the two tracks. The encoder state moves here, so the VP9 picture ids + /// continue. The SFU drops a stream whose timestamps go back. The + /// timestamps also count the time since `previous` stopped. fn continue_rtp_from(&self, previous: &LocalVideoTrack) { let stopped_for = previous.inner.encodings[0] .core .stopped_at .get() .map_or(Duration::ZERO, Instant::elapsed); - let samples = - (stopped_for.as_secs_f64() * f64::from(self.inner.encodings[0].core.clock_rate)) as u32; - 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 next_pts = previous + .inner + .clock + .lock() + .unwrap_or_else(|e| e.into_inner()) + .next_pts; { let mut clock = self.inner.clock.lock().unwrap_or_else(|e| e.into_inner()); clock.next_pts = next_pts; - clock.rtp_ts = rtp_ts; - clock.advance( - i64::try_from(stopped_for.as_millis()).unwrap_or(i64::MAX), - samples, - ); + clock.advance(i64::try_from(stopped_for.as_millis()).unwrap_or(i64::MAX)); } let (current, previous) = (&self.inner.encodings[0], &previous.inner.encodings[0]); let state = previous @@ -1395,18 +1449,7 @@ impl LocalVideoTrack { .unwrap_or_else(|e| e.into_inner()) .take(); *current.encoder.lock().unwrap_or_else(|e| e.into_inner()) = state; - let mut packetizer = previous - .core - .packetizer - .lock() - .unwrap_or_else(|e| e.into_inner()) - .clone(); - packetizer.skip_samples(samples); - *current - .core - .packetizer - .lock() - .unwrap_or_else(|e| e.into_inner()) = packetizer; + current.core.continue_rtp_from(&previous.core); } /// The underlying webrtc-rs track, for attaching this track to a @@ -1779,11 +1822,17 @@ fn encode_i420_layers( dur_ms: i64, samples: u32, ) -> Result)>> { - let (pts, timestamp) = inner + let pts = inner .clock .lock() .unwrap_or_else(|error| error.into_inner()) - .advance(dur_ms, samples); + .advance(dur_ms); + let timestamp = inner.encodings[0] + .core + .rtp + .lock() + .unwrap_or_else(|error| error.into_inner()) + .advance(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() { @@ -1853,10 +1902,6 @@ fn encode_layer_packets( None => true, }; if needs_new { - let seq = guard - .as_ref() - .map(|state| state.seq) - .unwrap_or_else(seed_u16); let prior_vp9_packetizer = guard.as_ref().and_then(|state| match &state.codec { VideoCodecState::Vp9Svc { packetizer, .. } => Some(packetizer.clone()), VideoCodecState::Vpx { .. } => None, @@ -1885,7 +1930,6 @@ fn encode_layer_packets( // Force a keyframe on the first frame after (re)init. last_key_pts: pts - KEYFRAME_INTERVAL_MS, last_emit_pts: None, - seq, }); } let Some(state) = guard.as_mut() else { @@ -1908,7 +1952,6 @@ fn encode_layer_packets( state.last_key_pts = pts; } let mut out = Vec::new(); - let mut seq = state.seq; let w16 = width.min(u32::from(u16::MAX)) as u16; let h16 = height.min(u32::from(u16::MAX)) as u16; match &mut state.codec { @@ -1931,7 +1974,6 @@ fn encode_layer_packets( let header = webrtc::rtp::header::Header { version: 2, payload_type: PLACEHOLDER_PT, - sequence_number: seq, timestamp, ssrc: PLACEHOLDER_SSRC, marker: payload.last, @@ -1941,7 +1983,6 @@ fn encode_layer_packets( header, payload: Bytes::from(payload.data), }); - seq = seq.wrapping_add(1); } } } @@ -1965,7 +2006,6 @@ fn encode_layer_packets( let header = webrtc::rtp::header::Header { version: 2, payload_type: PLACEHOLDER_PT, - sequence_number: seq, timestamp, ssrc: PLACEHOLDER_SSRC, marker: payload.last, @@ -1975,11 +2015,13 @@ fn encode_layer_packets( header, payload: Bytes::from(payload.data), }); - seq = seq.wrapping_add(1); } } } - state.seq = seq; + let mut rtp = encoding.core.rtp.lock().unwrap_or_else(|e| e.into_inner()); + for packet in &mut out { + packet.header.sequence_number = rtp.next_seq(); + } Ok(out) } From 24672822f8d887ebfb0acf0ed8149582866a8a0a Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 12 Oct 2026 15:33:25 +0200 Subject: [PATCH 4/7] docs: add the write_rtp takeover to the changelog --- CHANGELOG.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b8768d..fb7d0b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -137,7 +137,9 @@ lost, and a reconnect adds no delay. 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 +media of the new track. The new track continues the RTP sequence numbers and +timestamps of the stopped track, also when one of the two tracks forwards RTP +with `write_rtp`. Before, a second video publish failed with "participant not found". A simulcast track (`vp8_simulcast`) cannot replace or be replaced, and a track cannot replace a track with different layers (for example `vp9_svc` and `vp9`): `publish` returns `RtcError::SimulcastReplace`. From 202a7c2e37e397b310e6f62b74363ee75114f703 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 12 Oct 2026 15:38:20 +0200 Subject: [PATCH 5/7] docs: ask agents to use the tdd skill for features and fixes --- AGENTS.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index b3b3fb1..b533254 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,6 +16,8 @@ UI are out of scope unless the task explicitly expands the project. - Never inspect, modify, or commit `.env`, credentials, generated secrets, or logs containing tokens. `.env.example` is the public configuration reference. +- Use /tdd skill when introducing new features or fixes to the codebase. + ## Rust and async expectations - Ensure [rust-skills](https://github.com/leonardomso/rust-skills) is available From 7cc8e99288483b53508623ab0699f7d4e098d750 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 12 Oct 2026 15:44:53 +0200 Subject: [PATCH 6/7] fix: count a write that comes during a pacer tick for the next tick take_pcm read pcm_written in the middle of a tick and cleared it at the end. A write that set the flag between the two was lost, and the next tick could flush the resampler while the stream continued. take_pcm now swaps the flag to false at the start of the tick. --- src/rtc/tracks/local.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index eda7c9b..503d185 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -460,6 +460,8 @@ impl AudioInner { /// While a write holds the resampler, the pacer does not wait and takes no /// such audio. fn take_pcm(&self, frame: &mut [i16]) { + // A write that sets the flag after this point counts for the next tick. + let written = self.pcm_written.swap(false, Ordering::SeqCst); let mut buf = self.pcm.lock().unwrap_or_else(|e| e.into_inner()); if buf.len() < frame.len() { drop(buf); @@ -470,7 +472,7 @@ impl AudioInner { }; buf = self.pcm.lock().unwrap_or_else(|e| e.into_inner()); if let Some(mut resampler) = resampler - && !self.pcm_written.load(Ordering::SeqCst) + && !written && buf.len() < frame.len() { match resampler.flush() { @@ -488,7 +490,6 @@ impl AudioInner { } } } - self.pcm_written.store(false, Ordering::SeqCst); for slot in frame.iter_mut() { *slot = buf.pop_front().unwrap_or(0); } From 20a3935da71b0a8cca3a06710aff92a7e5b77f95 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 12 Oct 2026 15:44:54 +0200 Subject: [PATCH 7/7] test: correct the pacer comment in the real-time PCM test The track has pace off, so no pacer starts. The test itself takes the frames. --- src/rtc/tracks/local.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index 503d185..fd273dd 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -2561,8 +2561,8 @@ mod tests { .write_pcm(PcmFrame::mono(samples.to_vec(), rate)) .await .expect("write_pcm"); - // The first write starts the pacer, and its first tick comes at - // once. Each later tick comes after two 10 ms writes. + // The test acts as the pacer: the first tick comes at once after + // the first write, and each later tick after two 10 ms writes. if index % 2 == 0 { track.inner.take_pcm(&mut frame); played.extend_from_slice(&frame);