diff --git a/CMakeLists.txt b/CMakeLists.txt index a4c3a6b..b1e526c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -408,6 +408,7 @@ add_library(livekit SHARED src/realtime_audio_processing.cpp src/audio_source.cpp src/audio_stream.cpp + src/fresh_audio_stream.cpp src/data_track_frame.cpp src/data_stream.cpp src/data_track_error.cpp diff --git a/docs/fresh-audio-stream.md b/docs/fresh-audio-stream.md new file mode 100644 index 0000000..53743ce --- /dev/null +++ b/docs/fresh-audio-stream.md @@ -0,0 +1,81 @@ +# Fresh native audio ingress + +`FreshAudioStream` is a track-bound input for native playout. It returns owned +48 kHz stereo PCM in 10 ms frames. It does not subscribe a publication, select an +output device, mix participants, or change Room state. Those remain application +owners. Stop all streams before disconnecting their Room or shutting down the SDK. + +```text +WebRTC decoded sink (timestamp before conversion) + -> NativeAudioSink format conversion + -> four preallocated PCM frames + -> synchronous private native read + -> FreshAudioFrame with steady-clock decoded_at +``` + +Construct, read, inspect stats, and destroy a stream on one owner thread. Reads +never wait for a frame. The queue admits at most four 960-sample values; overflow +evicts the oldest. Producer and consumer each make one atomic attempt to access +the queue. Contention drops the incoming frame or returns `empty` to the reader. +The queue does not allocate, log, notify a worker, serialize protobuf, or invoke +an application callback. It bypasses the legacy Rust and C++ `AudioStream` queues. + +Keep `FreshAudioFrame::decoded_at` unchanged through subsequent application +queues. It is in `std::chrono::steady_clock`'s domain. Conversion time, native +queue residence, and native-to-C++ bridge time contribute to age. Translation +uses the start of the native read and may conservatively overestimate age by +the duration of that call; it never intentionally refreshes the timestamp at +delivery. `age_us` is a snapshot at return, not a replacement timestamp. + +The default maximum age is 60 ms. A caller may request 1–1,000,000 microseconds; +larger buffers or age limits should not be used to hide a slow playout consumer. +Stale frames are discarded before delivery. `sequence`, `discontinuity`, and +the received/drop/stale/invalid/queued counters make gaps and capacity visible. +An empty read leaves its output value unchanged: callers must inspect the result +and must not replay their previous frame. + +The existing native sink performs rate/channel conversion and may initialize +its resampler when the input format changes. That work precedes the fixed queue +and is included in frame age; this API does not claim that all WebRTC internals +are allocation-free. Invalid converted frame lengths are counted and rejected. +There is no silent fallback to a different queue or an unbounded path. + +The stream retains its typed track. Destruction first retires ingress, then +removes the native sink before releasing that track. The application still +needs a publication/source generation fence: retaining a track does not grant +permission to play PCM after a participant's track is removed or replaced. + +With manual subscriptions, seed the application's publication registry after +connect using `Room::remotePublications(maximum)`. Initial publications do not +emit `onTrackPublished`. This bounded snapshot copies owning handles under the +Room lock, avoiding concurrent iteration over a participant's publication map. +It reports truncation and caps capacity at 4096 entries. Take snapshots on a +control thread, and fence their application against intervening Room events; +an owning handle alone does not make a removed publication current again. + +Verification lives in the Rust `fresh_audio_stream` tests and C++ +`AudioSourceTest.FreshSink*` tests. Full remote Room, output, latency, and resource +proofs are the consuming application's Media Lab responsibility. + +The deterministic Rust callback test injects 250 ms of native processing age +for 100 consecutive frames and verifies that none becomes fresh at the queue +boundary. Separate application probes delay the reader by 250 ms and exercise +renderer wake stalls. A 30-minute, two-participant application run measured +p95 scheduled age at most 50 ms and maximum 56.911 ms, with bounded queues and +no underruns; these are development measurements on one Windows machine. + +## Publisher replacement regression + +The remote-audio routing fixture uncovered a publisher regression with mixed +microphone, screen audio and screen video: after a track replacement worked, +removing a different track could also remove that replacement at the receiver. +Disabling transceiver reuse removed the symptom, but would discard the existing +resource bound. Waiting for removal offer/answer settlement did not fix it. + +The correction keeps libwebrtc's sender MSID in `SetLocalDescription` and applies +the current track identity only to the offer emitted to the SFU. All codec, +MID, SSRC and transport fields remain the same. The fixture passes 17 phases, +including immediate screen replacement, microphone replacement, adjacent source +continuity, demand fencing, persistent volume and receiver Room replacement. +A native Rust regression test checks local/wire identities over replacement and +a subsequent unrelated negotiation while retaining one RTP transceiver. diff --git a/include/livekit/fresh_audio_stream.h b/include/livekit/fresh_audio_stream.h new file mode 100644 index 0000000..296a72a --- /dev/null +++ b/include/livekit/fresh_audio_stream.h @@ -0,0 +1,83 @@ +/* + * Copyright 2026 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include + +#include "livekit/track.h" +#include "livekit/visibility.h" + +namespace livekit { + +struct FreshAudioFrame { + std::array samples{}; + std::uint64_t sequence = 0; + /// Elapsed monotonic time since decoded native sink ingress, before queues. + std::uint64_t age_us = 0; + /// Conservative timestamp in the consuming C++ process's steady-clock domain. + /// Preserve this value through application queues instead of re-stamping PCM. + std::chrono::steady_clock::time_point decoded_at{}; + bool discontinuity = false; +}; + +struct FreshAudioStreamStats { + std::uint64_t received = 0; + std::uint64_t dropped = 0; + std::uint64_t stale = 0; + std::uint64_t invalid = 0; + std::uint64_t queued = 0; +}; + +enum class FreshAudioReadResult { frame, empty, invalid_input, wrong_thread }; + +/// @brief Track-bound 48 kHz stereo/10 ms decoded ingress for native playout. +/// Four preallocated frames; overload drops oldest, contention never waits. +/// No per-frame FFI event, callback thread, or legacy AudioStream queue. +/// Construct/read/destroy on one owner, before Room/SDK teardown. Copy the +/// returned decoded_at through the application's own queues unchanged. +/// NativeAudioSink format conversion is included in the age. It may initialize +/// its resampler on a format change; this API bounds the post-conversion queue. +class LIVEKIT_API FreshAudioStream final { +public: + explicit FreshAudioStream(std::shared_ptr track); + ~FreshAudioStream(); + FreshAudioStream(const FreshAudioStream&) = delete; + FreshAudioStream& operator=(const FreshAudioStream&) = delete; + FreshAudioStream(FreshAudioStream&&) = delete; + FreshAudioStream& operator=(FreshAudioStream&&) = delete; + + /// Read once without waiting. Maximum age must be 1..1,000,000 microseconds. + /// On empty/error, output is unchanged. Stats are refreshed by valid reads. + [[nodiscard]] FreshAudioReadResult tryRead(FreshAudioFrame& output, std::uint64_t maximum_age_us = 60'000) noexcept; + [[nodiscard]] FreshAudioStreamStats stats() const noexcept { return stats_; } + +private: + struct Deleter { + void operator()(void* value) const noexcept; + }; + std::shared_ptr track_; + std::unique_ptr state_; + const std::thread::id owner_ = std::this_thread::get_id(); + FreshAudioStreamStats stats_; + std::uint64_t bridge_stale_ = 0; + std::uint64_t delivered_sequence_ = 0; +}; +} // namespace livekit diff --git a/include/livekit/room.h b/include/livekit/room.h index 5eda88a..ab4e552 100644 --- a/include/livekit/room.h +++ b/include/livekit/room.h @@ -43,6 +43,15 @@ class E2EEManager; class LocalParticipant; class RemoteParticipant; +struct RemotePublicationSnapshot { + struct Entry { + std::shared_ptr participant; + std::shared_ptr publication; + }; + std::vector entries; + bool truncated = false; +}; + /// Represents a single ICE server configuration. struct IceServer { /// TURN/STUN server URL (e.g. "stun:stun.l.google.com:19302"). @@ -237,6 +246,12 @@ class LIVEKIT_API Room { /// corresponding participant disconnects or the room is torn down. std::vector> remoteParticipants() const; + /// Copies publication handles under the Room lock, including publications + /// present before connect (which do not emit onTrackPublished). Use on a + /// control thread. Capacity is capped at 4096; truncation is explicit. + /// Handles own their lifetimes, but later Room events can supersede them. + [[nodiscard]] RemotePublicationSnapshot remotePublications(std::size_t maximum) const; + /// Returns the current connection state of the room. ConnectionState connectionState() const; diff --git a/patches/client-sdk-rust/0001-native-v2-runtime.patch b/patches/client-sdk-rust/0001-native-v2-runtime.patch index 7816739..479ce8c 100644 --- a/patches/client-sdk-rust/0001-native-v2-runtime.patch +++ b/patches/client-sdk-rust/0001-native-v2-runtime.patch @@ -1,3 +1,18 @@ +diff --git a/.changeset/fresh-audio-ingress.md b/.changeset/fresh-audio-ingress.md +new file mode 100644 +index 00000000..8c07b23f +--- /dev/null ++++ b/.changeset/fresh-audio-ingress.md +@@ -0,0 +1,9 @@ ++--- ++livekit-ffi: patch ++libwebrtc: patch ++webrtc-sys: patch ++--- ++ ++Add a fixed-capacity native decoded audio ingress with monotonic frame age, ++nonblocking access, explicit drop counters and deterministic track-bound ++ownership for fresh playout. The existing asynchronous AudioStream API is unchanged. diff --git a/.changeset/microphone-realtime-apm.md b/.changeset/microphone-realtime-apm.md new file mode 100644 index 00000000..aea85aa7 @@ -76,6 +91,20 @@ index 00000000..d267a689 +The pass-through encoder requests a fresh keyframe and suppresses dependent +access units until it arrives, instead of forwarding a damaged reference chain. +Continuity is private to each encoder and resets on encoder/source replacement. +diff --git a/.changeset/publisher-wire-msid.md b/.changeset/publisher-wire-msid.md +new file mode 100644 +index 00000000..9a1363bb +--- /dev/null ++++ b/.changeset/publisher-wire-msid.md +@@ -0,0 +1,8 @@ ++--- ++"livekit": patch ++--- ++ ++Keep the immutable libwebrtc sender MSID in the local SDP when reusing a ++publisher transceiver. Apply the current AddTrack identity only to the offer ++sent to the SFU, so a later unrelated negotiation does not remove the replacement ++publication. Transceiver reuse and its resource bound remain enabled. diff --git a/.changeset/reused-screen-sender.md b/.changeset/reused-screen-sender.md new file mode 100644 index 00000000..5896d522 @@ -93,6 +122,20 @@ index 00000000..5896d522 +replacement screen track. Return an absent track safely for inactive senders. +This preserves bounded transceiver reuse while allowing the SFU to associate +new RTP media with the replacement publication. +diff --git a/libwebrtc/src/lib.rs b/libwebrtc/src/lib.rs +index 97611978..ee40bf25 100644 +--- a/libwebrtc/src/lib.rs ++++ b/libwebrtc/src/lib.rs +@@ -71,7 +71,8 @@ pub mod native { + pub use webrtc_sys::webrtc::ffi::create_random_uuid; + + pub use crate::imp::{ +- apm, audio_mixer, audio_resampler, frame_cryptor, packet_trailer, yuv_helper, ++ apm, audio_mixer, audio_resampler, frame_cryptor, fresh_audio_stream, packet_trailer, ++ yuv_helper, + }; + } + diff --git a/libwebrtc/src/native/apm.rs b/libwebrtc/src/native/apm.rs index a5c53fb9..f40b204d 100644 --- a/libwebrtc/src/native/apm.rs @@ -136,6 +179,396 @@ index a5c53fb9..f40b204d 100644 pub fn new( echo_canceller_enabled: bool, gain_controller_enabled: bool, +diff --git a/libwebrtc/src/native/fresh_audio_stream.rs b/libwebrtc/src/native/fresh_audio_stream.rs +new file mode 100644 +index 00000000..5d90f21d +--- /dev/null ++++ b/libwebrtc/src/native/fresh_audio_stream.rs +@@ -0,0 +1,372 @@ ++// Copyright 2026 LiveKit, Inc. ++// ++// Licensed under the Apache License, Version 2.0 (the "License"); ++// you may not use this file except in compliance with the License. ++// You may obtain a copy of the License at ++// ++// http://www.apache.org/licenses/LICENSE-2.0 ++// ++// Unless required by applicable law or agreed to in writing, software ++// distributed under the License is distributed on an "AS IS" BASIS, ++// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. ++// See the License for the specific language governing permissions and ++// limitations under the License. ++ ++//! Fixed decoded ingress for native playout. No Tokio task, FFI event queue, ++//! per-frame allocation, receiver waker, or waiting lock in this handoff. ++ ++use std::{ ++ cell::UnsafeCell, ++ ops::{Deref, DerefMut}, ++ sync::{ ++ atomic::{AtomicBool, AtomicU64, Ordering}, ++ Arc, ++ }, ++ time::{Duration, Instant}, ++}; ++ ++use cxx::SharedPtr; ++use webrtc_sys::audio_track as sys_at; ++ ++use crate::audio_track::RtcAudioTrack; ++ ++/// Samples in one interleaved 48 kHz stereo/10 ms frame. ++pub const FRESH_AUDIO_SAMPLES: usize = 960; ++/// Hard number of retained decoded frames per sink. ++pub const FRESH_AUDIO_CAPACITY: usize = 4; ++ ++/// Owned fixed PCM and the monotonic instant preceding its ingress copy. ++#[derive(Clone, Copy)] ++pub struct FreshAudioFrame { ++ pub samples: [i16; FRESH_AUDIO_SAMPLES], ++ pub decoded_at: Instant, ++ pub sequence: u64, ++ pub discontinuity: bool, ++} ++ ++/// Bounded ingress counters; no sample-level telemetry callback is emitted. ++#[derive(Clone, Copy, Default)] ++pub struct FreshAudioStats { ++ pub received: u64, ++ pub dropped: u64, ++ pub stale: u64, ++ pub invalid: u64, ++ pub queued: u64, ++} ++ ++struct QueueState { ++ frames: [Option; FRESH_AUDIO_CAPACITY], ++ head: usize, ++ depth: usize, ++ consumed_sequence: u64, ++} ++ ++struct FreshAudioQueue { ++ state: UnsafeCell, ++ busy: AtomicBool, ++ closed: AtomicBool, ++ received: AtomicU64, ++ dropped: AtomicU64, ++ stale: AtomicU64, ++ invalid: AtomicU64, ++ queued: AtomicU64, ++} ++ ++// SAFETY: Every access to state requires the single successful acquire CAS in ++// try_state. QueueGuard exclusively borrows it until its release store. No ++// reference into state escapes a guard; frames are copied out as owned values. ++unsafe impl Sync for FreshAudioQueue {} ++ ++struct QueueGuard<'a>(&'a FreshAudioQueue); ++ ++impl Deref for QueueGuard<'_> { ++ type Target = QueueState; ++ fn deref(&self) -> &Self::Target { ++ // SAFETY: This guard owns the queue's one state access permit. ++ unsafe { &*self.0.state.get() } ++ } ++} ++ ++impl DerefMut for QueueGuard<'_> { ++ fn deref_mut(&mut self) -> &mut Self::Target { ++ // SAFETY: This unique guard owns the queue's one state access permit. ++ unsafe { &mut *self.0.state.get() } ++ } ++} ++ ++impl Drop for QueueGuard<'_> { ++ fn drop(&mut self) { ++ self.0.busy.store(false, Ordering::Release); ++ } ++} ++ ++impl FreshAudioQueue { ++ fn new() -> Self { ++ Self { ++ state: UnsafeCell::new(QueueState { ++ frames: [None; FRESH_AUDIO_CAPACITY], ++ head: 0, ++ depth: 0, ++ consumed_sequence: 0, ++ }), ++ busy: AtomicBool::new(false), ++ closed: AtomicBool::new(false), ++ received: AtomicU64::new(0), ++ dropped: AtomicU64::new(0), ++ stale: AtomicU64::new(0), ++ invalid: AtomicU64::new(0), ++ queued: AtomicU64::new(0), ++ } ++ } ++ ++ fn try_state(&self) -> Option> { ++ self.busy ++ .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed) ++ .ok() ++ .map(|_| QueueGuard(self)) ++ } ++ ++ fn push(&self, data: &[i16], decoded_at: Instant) { ++ if self.closed.load(Ordering::Acquire) { ++ return; ++ } ++ if data.len() != FRESH_AUDIO_SAMPLES { ++ self.invalid.fetch_add(1, Ordering::Relaxed); ++ return; ++ } ++ let sequence = self.received.fetch_add(1, Ordering::Relaxed) + 1; ++ let Some(mut state) = self.try_state() else { ++ self.dropped.fetch_add(1, Ordering::Relaxed); ++ return; ++ }; ++ if self.closed.load(Ordering::Acquire) { ++ return; ++ } ++ if state.depth == FRESH_AUDIO_CAPACITY { ++ state.head = (state.head + 1) % FRESH_AUDIO_CAPACITY; ++ state.depth -= 1; ++ self.dropped.fetch_add(1, Ordering::Relaxed); ++ } ++ let mut frame = FreshAudioFrame { ++ samples: [0; FRESH_AUDIO_SAMPLES], ++ decoded_at, ++ sequence, ++ discontinuity: false, ++ }; ++ frame.samples.copy_from_slice(data); ++ let index = (state.head + state.depth) % FRESH_AUDIO_CAPACITY; ++ state.frames[index] = Some(frame); ++ state.depth += 1; ++ self.queued.store(state.depth as u64, Ordering::Relaxed); ++ } ++ ++ fn take(&self, now: Instant, maximum_age: Duration) -> Option { ++ if self.closed.load(Ordering::Acquire) { ++ return None; ++ } ++ let mut state = self.try_state()?; ++ for _ in 0..FRESH_AUDIO_CAPACITY { ++ if state.depth == 0 { ++ break; ++ } ++ let index = state.head; ++ let frame = state.frames[index].take(); ++ state.head = (state.head + 1) % FRESH_AUDIO_CAPACITY; ++ state.depth -= 1; ++ self.queued.store(state.depth as u64, Ordering::Relaxed); ++ if let Some(mut frame) = frame { ++ if now < frame.decoded_at || now.duration_since(frame.decoded_at) > maximum_age { ++ self.stale.fetch_add(1, Ordering::Relaxed); ++ continue; ++ } ++ frame.discontinuity = frame.sequence != state.consumed_sequence + 1; ++ state.consumed_sequence = frame.sequence; ++ return Some(frame); ++ } ++ } ++ None ++ } ++ ++ fn stats(&self) -> FreshAudioStats { ++ FreshAudioStats { ++ received: self.received.load(Ordering::Relaxed), ++ dropped: self.dropped.load(Ordering::Relaxed), ++ stale: self.stale.load(Ordering::Relaxed), ++ invalid: self.invalid.load(Ordering::Relaxed), ++ queued: if self.closed.load(Ordering::Acquire) { ++ 0 ++ } else { ++ self.queued.load(Ordering::Relaxed) ++ }, ++ } ++ } ++} ++ ++impl sys_at::AudioSink for FreshAudioQueue { ++ fn on_data(&self, data: &[i16], sample_rate: i32, channels: usize, frames: usize) { ++ self.on_data_with_age(data, sample_rate, channels, frames, 0); ++ } ++ ++ fn on_data_with_age( ++ &self, ++ data: &[i16], ++ sample_rate: i32, ++ channels: usize, ++ frames: usize, ++ processing_age_us: u64, ++ ) { ++ // Translate elapsed time rather than assuming C++ and Rust Instant ++ // share an epoch. The native age includes format conversion/scheduling. ++ let Some(decoded_at) = Instant::now().checked_sub(Duration::from_micros(processing_age_us)) ++ else { ++ self.invalid.fetch_add(1, Ordering::Relaxed); ++ return; ++ }; ++ if sample_rate != 48_000 || channels != 2 || frames != 480 { ++ self.invalid.fetch_add(1, Ordering::Relaxed); ++ return; ++ } ++ self.push(data, decoded_at); ++ } ++} ++ ++/// One track-bound native sink. Create/read/close on one owner; WebRTC owns the ++/// producer callback. Stop this stream before closing its Room/runtime. ++pub struct FreshAudioStream { ++ sink: SharedPtr, ++ track: RtcAudioTrack, ++ queue: Arc, ++ closed: bool, ++} ++ ++impl FreshAudioStream { ++ /// Attach one converted 48 kHz stereo native sink to the audio track. ++ pub fn new(track: RtcAudioTrack) -> Self { ++ let queue = Arc::new(FreshAudioQueue::new()); ++ let sink = sys_at::ffi::new_native_audio_sink( ++ Box::new(sys_at::AudioSinkWrapper::new(queue.clone())), ++ 48_000, ++ 2, ++ ); ++ // SAFETY: RtcAudioTrack guarantees that this handle is an audio track. ++ let audio = unsafe { sys_at::ffi::media_to_audio(track.sys_handle()) }; ++ audio.add_sink(&sink); ++ Self { sink, track, queue, closed: false } ++ } ++ ++ /// Copy one fresh frame without waiting; discard expired queued values. ++ pub fn try_read(&self, maximum_age: Duration) -> Option { ++ self.queue.take(Instant::now(), maximum_age) ++ } ++ ++ /// Snapshot counters without entering the PCM critical section. ++ pub fn stats(&self) -> FreshAudioStats { ++ self.queue.stats() ++ } ++ ++ /// Retire ingress before removing the native sink; repeated close is safe. ++ pub fn close(&mut self) { ++ if self.closed { ++ return; ++ } ++ self.closed = true; ++ self.queue.closed.store(true, Ordering::Release); ++ // SAFETY: The retained typed track keeps its native audio handle live. ++ let audio = unsafe { sys_at::ffi::media_to_audio(self.track.sys_handle()) }; ++ audio.remove_sink(&self.sink); ++ } ++} ++ ++impl Drop for FreshAudioStream { ++ fn drop(&mut self) { ++ self.close(); ++ } ++} ++ ++#[cfg(test)] ++mod tests { ++ use super::*; ++ ++ #[test] ++ fn overrun_stale_and_retirement() { ++ let queue = FreshAudioQueue::new(); ++ let now = Instant::now(); ++ for index in 0..10 { ++ queue.push(&[index; FRESH_AUDIO_SAMPLES], now); ++ } ++ assert_eq!(queue.stats().queued, 4); ++ assert_eq!(queue.stats().dropped, 6); ++ let frame = queue.take(now, Duration::from_millis(60)).unwrap(); ++ assert_eq!(frame.samples[0], 6); ++ assert!(frame.discontinuity); ++ assert!(queue.take(now + Duration::from_secs(1), Duration::from_millis(60)).is_none()); ++ assert_eq!(queue.stats().stale, 3); ++ queue.closed.store(true, Ordering::Release); ++ queue.push(&[1; FRESH_AUDIO_SAMPLES], now); ++ assert_eq!(queue.stats().received, 10); ++ assert!(queue.take(now, Duration::from_secs(1)).is_none()); ++ } ++ ++ #[test] ++ fn callback_drops_on_contention_without_waiting() { ++ let queue = FreshAudioQueue::new(); ++ let guard = queue.try_state().unwrap(); ++ queue.push(&[1; FRESH_AUDIO_SAMPLES], Instant::now()); ++ assert_eq!(queue.stats().dropped, 1); ++ assert!(queue.take(Instant::now(), Duration::from_millis(60)).is_none()); ++ drop(guard); ++ queue.push(&[2; FRESH_AUDIO_SAMPLES], Instant::now()); ++ assert!(queue.take(Instant::now(), Duration::from_millis(60)).unwrap().discontinuity); ++ } ++ ++ #[test] ++ fn slow_native_callback_does_not_refresh_decoded_age() { ++ use sys_at::AudioSink; ++ ++ let queue = FreshAudioQueue::new(); ++ for _ in 0..100 { ++ // Inject the elapsed conversion/callback time supplied by the native ++ // bridge, independently of time spent waiting in the reader queue. ++ queue.on_data_with_age(&[1; FRESH_AUDIO_SAMPLES], 48_000, 2, 480, 250_000); ++ assert!(queue.take(Instant::now(), Duration::from_millis(60)).is_none()); ++ } ++ assert_eq!(queue.stats().stale, 100); ++ assert_eq!(queue.stats().queued, 0); ++ queue.on_data_with_age(&[2; FRESH_AUDIO_SAMPLES], 48_000, 2, 480, 0); ++ let recovered = queue.take(Instant::now(), Duration::from_secs(1)).unwrap(); ++ assert_eq!(recovered.samples[0], 2); ++ assert!(recovered.discontinuity); ++ } ++ ++ #[test] ++ fn concurrent_frames_are_owned_and_ordered() { ++ let queue = Arc::new(FreshAudioQueue::new()); ++ let done = Arc::new(AtomicBool::new(false)); ++ let now = Instant::now(); ++ let producer_queue = queue.clone(); ++ let producer_done = done.clone(); ++ let producer = std::thread::spawn(move || { ++ for sequence in 1..=100_000u64 { ++ producer_queue.push(&[(sequence % 30_000) as i16; FRESH_AUDIO_SAMPLES], now); ++ } ++ producer_done.store(true, Ordering::Release); ++ }); ++ let mut last_sequence = 0; ++ let mut torn = false; ++ while !done.load(Ordering::Acquire) || queue.stats().queued != 0 { ++ if let Some(frame) = queue.take(now, Duration::from_secs(1)) { ++ torn |= frame.sequence <= last_sequence; ++ torn |= ++ frame.samples.iter().any(|sample| *sample != (frame.sequence % 30_000) as i16); ++ last_sequence = frame.sequence; ++ } else { ++ std::thread::yield_now(); ++ } ++ } ++ producer.join().unwrap(); ++ assert!(!torn); ++ assert!(last_sequence != 0); ++ assert_eq!(queue.stats().received, 100_000); ++ assert_eq!(queue.stats().queued, 0); ++ } ++} +diff --git a/libwebrtc/src/native/mod.rs b/libwebrtc/src/native/mod.rs +index de56e334..f74a1788 100644 +--- a/libwebrtc/src/native/mod.rs ++++ b/libwebrtc/src/native/mod.rs +@@ -24,6 +24,7 @@ pub mod data_channel; + #[cfg(any(target_os = "macos", target_os = "windows", target_os = "linux"))] + pub mod desktop_capturer; + pub mod frame_cryptor; ++pub mod fresh_audio_stream; + pub mod ice_candidate; + pub mod media_stream; + pub mod media_stream_track; diff --git a/libwebrtc/src/native/rtp_transceiver.rs b/libwebrtc/src/native/rtp_transceiver.rs index 3bd4fb59..a6a089bc 100644 --- a/libwebrtc/src/native/rtp_transceiver.rs @@ -318,14 +751,135 @@ index 4922add5..281dcfc7 100644 ]; for (proto_backend, expected) in cases { +diff --git a/livekit-ffi/src/fresh_audio.rs b/livekit-ffi/src/fresh_audio.rs +new file mode 100644 +index 00000000..2d1f2010 +--- /dev/null ++++ b/livekit-ffi/src/fresh_audio.rs +@@ -0,0 +1,114 @@ ++// Copyright 2026 LiveKit, Inc. ++// ++// Licensed under the Apache License, Version 2.0 (the "License"); ++// you may not use this file except in compliance with the License. ++// You may obtain a copy of the License at ++// ++// http://www.apache.org/licenses/LICENSE-2.0 ++// ++// Unless required by applicable law or agreed to in writing, software ++// distributed under the License is distributed on an "AS IS" BASIS, ++// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. ++// See the License for the specific language governing permissions and ++// limitations under the License. ++ ++//! Private fixed-frame native ABI. The public C++ class owns these pointers; ++//! no per-frame protobuf allocation, asynchronous dispatch, or handle lookup. ++ ++use livekit::webrtc::{ ++ native::fresh_audio_stream::{FreshAudioStream, FRESH_AUDIO_SAMPLES}, ++ prelude::MediaStreamTrack, ++}; ++use std::{ffi::c_void, panic, ptr, time::Duration}; ++ ++use crate::{server::room::FfiTrack, FFI_SERVER}; ++ ++#[repr(C)] ++/// Private C ABI result containing frame age and bounded ingress counters. ++pub struct FreshAudioInfo { ++ sequence: u64, ++ age_us: u64, ++ received: u64, ++ dropped: u64, ++ stale: u64, ++ invalid: u64, ++ queued: u64, ++ discontinuity: u32, ++} ++ ++/// Create a sink for a live audio track. Returns null for invalid/non-audio IDs. ++#[no_mangle] ++pub extern "C" fn livekit_ffi_fresh_audio_create(track_handle: u64) -> *mut c_void { ++ panic::catch_unwind(|| { ++ let Ok(track) = FFI_SERVER.retrieve_handle::(track_handle) else { ++ return ptr::null_mut(); ++ }; ++ let MediaStreamTrack::Audio(audio) = track.track.rtc_track() else { ++ return ptr::null_mut(); ++ }; ++ Box::into_raw(Box::new(FreshAudioStream::new(audio))).cast() ++ }) ++ .unwrap_or(ptr::null_mut()) ++} ++ ++/// # Safety ++/// Handle is null or the exclusively owned live create result. Call once, ++/// after all reads, and before Room/SDK destruction. No concurrent calls. ++#[no_mangle] ++pub unsafe extern "C" fn livekit_ffi_fresh_audio_destroy(handle: *mut c_void) { ++ if !handle.is_null() { ++ // SAFETY: The public owner returns exclusive ownership once. ++ drop(unsafe { Box::from_raw(handle.cast::()) }); ++ } ++} ++ ++/// Return 1 for a frame, 0 for empty/contention/stale, -1 for invalid arguments. ++/// Stats are returned even when the queue is empty; PCM is untouched then. ++/// ++/// # Safety ++/// Handle is a live create result. Samples is a writable array of exactly 960 ++/// i16, info is a writable FreshAudioInfo; both are aligned and do not alias the ++/// handle or each other. One owner serializes reads/destruction. ++#[no_mangle] ++pub unsafe extern "C" fn livekit_ffi_fresh_audio_read( ++ handle: *mut c_void, ++ samples: *mut i16, ++ sample_count: usize, ++ maximum_age_us: u64, ++ info: *mut FreshAudioInfo, ++) -> i32 { ++ if handle.is_null() ++ || samples.is_null() ++ || info.is_null() ++ || sample_count != FRESH_AUDIO_SAMPLES ++ || maximum_age_us == 0 ++ || maximum_age_us > 1_000_000 ++ { ++ return -1; ++ } ++ // SAFETY: All pointers follow the documented exclusive storage contract. ++ let stream = unsafe { &*handle.cast::() }; ++ let frame = stream.try_read(Duration::from_micros(maximum_age_us)); ++ let stats = stream.stats(); ++ // SAFETY: The caller supplies exclusive writable, aligned result storage. ++ let output = unsafe { &mut *info }; ++ *output = FreshAudioInfo { ++ sequence: 0, ++ age_us: 0, ++ received: stats.received, ++ dropped: stats.dropped, ++ stale: stats.stale, ++ invalid: stats.invalid, ++ queued: stats.queued, ++ discontinuity: 0, ++ }; ++ if let Some(frame) = frame { ++ output.sequence = frame.sequence; ++ output.age_us = frame.decoded_at.elapsed().as_micros().min(u64::MAX as u128) as u64; ++ output.discontinuity = u32::from(frame.discontinuity); ++ // SAFETY: The stack frame and caller's 960-sample output do not overlap. ++ unsafe { ptr::copy_nonoverlapping(frame.samples.as_ptr(), samples, FRESH_AUDIO_SAMPLES) }; ++ return 1; ++ } ++ 0 ++} diff --git a/livekit-ffi/src/lib.rs b/livekit-ffi/src/lib.rs -index 06695660..9c71e3fd 100644 +index 06695660..87cd8850 100644 --- a/livekit-ffi/src/lib.rs +++ b/livekit-ffi/src/lib.rs -@@ -19,6 +19,7 @@ use livekit::prelude::*; +@@ -19,6 +19,8 @@ use livekit::prelude::*; use thiserror::Error; mod conversion; ++mod fresh_audio; +mod realtime_apm; pub mod cabi; @@ -679,7 +1233,7 @@ index b02d88e5..c00819f8 100644 == inner.local_participant.identity().as_str() { diff --git a/livekit/src/rtc_engine/peer_transport.rs b/livekit/src/rtc_engine/peer_transport.rs -index 7eccd2ef..bff336e5 100644 +index 7eccd2ef..0e41446d 100644 --- a/livekit/src/rtc_engine/peer_transport.rs +++ b/livekit/src/rtc_engine/peer_transport.rs @@ -33,6 +33,7 @@ struct TransportInner { @@ -799,7 +1353,7 @@ index 7eccd2ef..bff336e5 100644 pub async fn create_and_send_offer(&self, options: OfferOptions) -> EngineResult<()> { let mut inner = self.inner.lock().await; if options.ice_restart { -@@ -478,6 +532,21 @@ impl PeerTransport { +@@ -478,6 +532,14 @@ impl PeerTransport { let mut offer = self.peer_connection.create_offer(options).await?; let mut sdp = offer.to_string(); @@ -811,17 +1365,10 @@ index 7eccd2ef..bff336e5 100644 + Some((transceiver.mid()?, transceiver.sender().track()?.id())) + }) + .collect(); -+ let current_ids = Self::munge_sender_track_ids(&sdp, &tracks); -+ if current_ids != sdp { -+ offer = SessionDescription::parse(¤t_ids, offer.sdp_type()).map_err(|error| { -+ super::EngineError::Internal(format!("Invalid current-track SDP: {error}").into()) -+ })?; -+ sdp = current_ids; -+ } if inner.single_pc_mode { // Fix inactive media m-lines to recvonly for single PC mode. -@@ -511,14 +580,18 @@ impl PeerTransport { +@@ -511,14 +573,18 @@ impl PeerTransport { } // Apply x-google-start-bitrate for all video codecs to improve initial quality. @@ -842,10 +1389,104 @@ index 7eccd2ef..bff336e5 100644 log::info!( "Applying x-google-start-bitrate={} kbps (target_bps={:?})", start_kbps, -@@ -552,8 +625,24 @@ impl PeerTransport { +@@ -542,6 +608,16 @@ impl PeerTransport { + + self.peer_connection.set_local_description(offer.clone()).await?; + ++ // Keep libwebrtc's immutable sender identity in its local description. ++ // The SFU's AddTrack identity belongs only to the signalled offer. ++ let local_sdp = offer.to_string(); ++ let current_ids = Self::munge_sender_track_ids(&local_sdp, &tracks); ++ if current_ids != local_sdp { ++ offer = SessionDescription::parse(¤t_ids, offer.sdp_type()).map_err(|error| { ++ super::EngineError::Internal(format!("Invalid current-track SDP: {error}").into()) ++ })?; ++ } ++ + if let Some(handler) = self.on_offer_handler.lock().as_mut() { + handler(offer); + } +@@ -552,8 +628,101 @@ impl PeerTransport { #[cfg(test)] mod tests { ++ #[cfg(not(target_arch = "wasm32"))] ++ #[tokio::test] ++ async fn reused_sender_keeps_local_identity_across_later_offers() { ++ use libwebrtc::{ ++ audio_source::native::NativeAudioSource, ++ peer_connection_factory::native::PeerConnectionFactoryExt, prelude::*, ++ }; ++ use parking_lot::Mutex; ++ use std::sync::Arc; ++ ++ let factory = PeerConnectionFactory::default(); ++ let publisher = factory.create_peer_connection(RtcConfiguration::default()).unwrap(); ++ let receiver = factory.create_peer_connection(RtcConfiguration::default()).unwrap(); ++ let source = NativeAudioSource::new(Default::default(), 48_000, 1, 10); ++ let original = factory.create_audio_track("original-sender", source.clone()); ++ let replacement = factory.create_audio_track("replacement-track", source); ++ let transceiver = publisher ++ .add_transceiver( ++ MediaStreamTrack::Audio(original), ++ RtpTransceiverInit { ++ direction: RtpTransceiverDirection::SendOnly, ++ stream_ids: Vec::new(), ++ send_encodings: Vec::new(), ++ }, ++ ) ++ .unwrap(); ++ let transport = ++ PeerTransport::new(publisher.clone(), livekit_protocol::SignalTarget::Publisher, false); ++ let emitted = Arc::new(Mutex::new(None)); ++ let output = emitted.clone(); ++ transport.on_offer(Some(Box::new(move |offer| *output.lock() = Some(offer)))); ++ let mut data_channel = None; ++ for phase in 0..3 { ++ if phase == 1 { ++ publisher.remove_track(transceiver.sender()).unwrap(); ++ transceiver ++ .sender() ++ .set_track(Some(MediaStreamTrack::Audio(replacement.clone()))) ++ .unwrap(); ++ transceiver.set_direction(RtpTransceiverDirection::SendOnly).unwrap(); ++ } ++ if phase == 2 { ++ // An unrelated subsequent negotiation exposed the regression: ++ // a replacement initially worked, then disappeared remotely. ++ data_channel = ++ Some(publisher.create_data_channel("unrelated", Default::default()).unwrap()); ++ } ++ tokio::time::timeout(std::time::Duration::from_secs(10), async { ++ transport.create_and_send_offer(OfferOptions::default()).await.unwrap(); ++ let offer = emitted.lock().take().expect("signalled offer"); ++ let expected = if phase == 0 { "original-sender" } else { "replacement-track" }; ++ assert!(offer ++ .to_string() ++ .lines() ++ .any(|line| line.starts_with("a=msid:") && line.ends_with(expected))); ++ receiver.set_remote_description(offer).await.unwrap(); ++ let answer = receiver.create_answer(AnswerOptions::default()).await.unwrap(); ++ receiver.set_local_description(answer.clone()).await.unwrap(); ++ transport.set_remote_description(answer).await.unwrap(); ++ let local = publisher.current_local_description().unwrap().to_string(); ++ assert!(local ++ .lines() ++ .any(|line| line.starts_with("a=msid:") && line.ends_with("original-sender"))); ++ assert_eq!( ++ publisher.transceivers().len(), ++ 1, ++ "replacement grew the RTP transceiver set" ++ ); ++ }) ++ .await ++ .expect("offer/answer deadline"); ++ } ++ drop(data_channel); ++ publisher.close(); ++ receiver.close(); ++ } ++ + #[test] + fn reused_sender_msid_follows_current_track_and_preserves_receiver() { + let original = "v=0\r\nm=video 9 RTP/AVP 96\r\na=mid:8\r\na=msid:- old\r\na=ssrc:123 msid:- old\r\na=ssrc:124 msid:screen old\r\nm=audio 9 RTP/AVP 111\r\na=mid:9\r\na=msid:remote receiver\r\n"; @@ -1652,6 +2293,102 @@ index 21337794..042ee055 100644 static std::shared_ptr video_to_media( std::shared_ptr track) { +diff --git a/webrtc-sys/src/audio_track.cpp b/webrtc-sys/src/audio_track.cpp +index 1738515f..e4f2c5ef 100644 +--- a/webrtc-sys/src/audio_track.cpp ++++ b/webrtc-sys/src/audio_track.cpp +@@ -17,6 +17,7 @@ + #include "livekit/audio_track.h" + + #include ++#include + #include + #include + #include +@@ -95,6 +96,11 @@ void NativeAudioSink::OnData(const void* audio_data, + int sample_rate, + size_t number_of_channels, + size_t number_of_frames) { ++ const auto decoded_at = std::chrono::steady_clock::now(); ++ const auto processing_age_us = [&] { ++ return static_cast(std::chrono::duration_cast( ++ std::chrono::steady_clock::now() - decoded_at).count()); ++ }; + RTC_CHECK_EQ(16, bits_per_sample); + + const int16_t* data = static_cast(audio_data); +@@ -110,14 +116,14 @@ void NativeAudioSink::OnData(const void* audio_data, + frame_.data(), frame_.num_channels() * frame_.samples_per_channel()); + + observer_->on_data(rust_slice, frame_.sample_rate_hz(), +- frame_.num_channels(), frame_.samples_per_channel()); ++ frame_.num_channels(), frame_.samples_per_channel(), processing_age_us()); + + } else { + rust::Slice rust_slice( + data, number_of_channels * number_of_frames); + + observer_->on_data(rust_slice, sample_rate, number_of_channels, +- number_of_frames); ++ number_of_frames, processing_age_us()); + } + } + +diff --git a/webrtc-sys/src/audio_track.rs b/webrtc-sys/src/audio_track.rs +index 67ed2701..23a3979d 100644 +--- a/webrtc-sys/src/audio_track.rs ++++ b/webrtc-sys/src/audio_track.rs +@@ -86,6 +86,7 @@ pub mod ffi { + sample_rate: i32, + nb_channels: usize, + nb_frames: usize, ++ processing_age_us: u64, + ); + } + } +@@ -107,6 +108,19 @@ unsafe impl ExternType for CompleteCallback { + + pub trait AudioSink: Send { + fn on_data(&self, data: &[i16], sample_rate: i32, nb_channels: usize, nb_frames: usize); ++ ++ /// Receive PCM with elapsed time since native decoded ingress, including ++ /// format conversion. Existing observers may ignore this added metadata. ++ fn on_data_with_age( ++ &self, ++ data: &[i16], ++ sample_rate: i32, ++ nb_channels: usize, ++ nb_frames: usize, ++ _processing_age_us: u64, ++ ) { ++ self.on_data(data, sample_rate, nb_channels, nb_frames); ++ } + } + + pub struct AudioSinkWrapper { +@@ -118,7 +132,20 @@ impl AudioSinkWrapper { + Self { observer } + } + +- fn on_data(&self, data: &[i16], sample_rate: i32, nb_channels: usize, nb_frames: usize) { +- self.observer.on_data(data, sample_rate, nb_channels, nb_frames); ++ fn on_data( ++ &self, ++ data: &[i16], ++ sample_rate: i32, ++ nb_channels: usize, ++ nb_frames: usize, ++ processing_age_us: u64, ++ ) { ++ self.observer.on_data_with_age( ++ data, ++ sample_rate, ++ nb_channels, ++ nb_frames, ++ processing_age_us, ++ ); + } + } diff --git a/webrtc-sys/src/passthrough_video_encoder.cpp b/webrtc-sys/src/passthrough_video_encoder.cpp index cd45b9e3..9d50e638 100644 --- a/webrtc-sys/src/passthrough_video_encoder.cpp diff --git a/src/fresh_audio_stream.cpp b/src/fresh_audio_stream.cpp new file mode 100644 index 0000000..81cb968 --- /dev/null +++ b/src/fresh_audio_stream.cpp @@ -0,0 +1,89 @@ +/* + * Copyright 2026 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "livekit/fresh_audio_stream.h" + +#include +#include +#include + +namespace { +struct NativeFreshAudioInfo { + std::uint64_t sequence; + std::uint64_t age_us; + std::uint64_t received; + std::uint64_t dropped; + std::uint64_t stale; + std::uint64_t invalid; + std::uint64_t queued; + std::uint32_t discontinuity; +}; +} // namespace + +extern "C" { +void* livekit_ffi_fresh_audio_create(std::uint64_t track_handle); +void livekit_ffi_fresh_audio_destroy(void* handle); +int livekit_ffi_fresh_audio_read(void* handle, std::int16_t* samples, std::size_t sample_count, + std::uint64_t maximum_age_us, NativeFreshAudioInfo* info); +} + +namespace livekit { +void FreshAudioStream::Deleter::operator()(void* value) const noexcept { livekit_ffi_fresh_audio_destroy(value); } + +FreshAudioStream::FreshAudioStream(std::shared_ptr track) : track_(std::move(track)) { + if (!track_ || track_->kind() != TrackKind::KIND_AUDIO) + throw std::invalid_argument("Fresh audio stream requires an audio track"); + state_.reset(livekit_ffi_fresh_audio_create(track_->ffiHandleId())); + if (!state_) throw std::runtime_error("Fresh audio stream initialization failed"); +} + +FreshAudioStream::~FreshAudioStream() { + if (owner_ != std::this_thread::get_id()) std::terminate(); +} + +FreshAudioReadResult FreshAudioStream::tryRead(FreshAudioFrame& output, std::uint64_t maximum_age_us) noexcept { + if (owner_ != std::this_thread::get_id()) return FreshAudioReadResult::wrong_thread; + if (!maximum_age_us || maximum_age_us > 1'000'000) return FreshAudioReadResult::invalid_input; + FreshAudioFrame next; + NativeFreshAudioInfo info{}; + const auto read_started = std::chrono::steady_clock::now(); + const auto result = + livekit_ffi_fresh_audio_read(state_.get(), next.samples.data(), next.samples.size(), maximum_age_us, &info); + if (result < 0) return FreshAudioReadResult::invalid_input; + stats_ = {info.received, info.dropped, info.stale + bridge_stale_, info.invalid, info.queued}; + if (!result) return FreshAudioReadResult::empty; + next.sequence = info.sequence; + if (info.age_us > static_cast((std::numeric_limits::max)())) + return FreshAudioReadResult::invalid_input; + // The Rust age is sampled during the native call. Subtract it from the call's + // START time: bridge time may conservatively increase age, never hide a stall. + next.decoded_at = read_started - std::chrono::microseconds{static_cast(info.age_us)}; + next.age_us = static_cast( + std::chrono::duration_cast(std::chrono::steady_clock::now() - next.decoded_at) + .count()); + if (next.age_us > maximum_age_us) { + ++bridge_stale_; + ++stats_.stale; + return FreshAudioReadResult::empty; + } + // Rust may already have consumed a frame that expired during the C++ bridge. + // Compare delivered sequences too, so that drop remains visible to callers. + next.discontinuity = info.discontinuity != 0 || next.sequence != delivered_sequence_ + 1; + delivered_sequence_ = next.sequence; + output = next; + return FreshAudioReadResult::frame; +} +} // namespace livekit diff --git a/src/room.cpp b/src/room.cpp index 57a6e3f..ca35754 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -363,6 +363,23 @@ ConnectionState Room::connectionState() const { return connection_state_; } +RemotePublicationSnapshot Room::remotePublications(std::size_t maximum) const { + maximum = std::min(maximum, std::size_t{4096}); + RemotePublicationSnapshot snapshot; + snapshot.entries.reserve(maximum); + const std::scoped_lock guard(lock_); + for (const auto& participant : remote_participants_) { + for (const auto& publication : participant.second->trackPublications()) { + if (snapshot.entries.size() == maximum) { + snapshot.truncated = true; + return snapshot; + } + snapshot.entries.push_back({participant.second, publication.second}); + } + } + return snapshot; +} + std::future Room::getStats() const { std::shared_ptr handle; { diff --git a/src/tests/unit/test_audio_source.cpp b/src/tests/unit/test_audio_source.cpp index d3a37a6..d7aae04 100644 --- a/src/tests/unit/test_audio_source.cpp +++ b/src/tests/unit/test_audio_source.cpp @@ -16,9 +16,14 @@ #include #include +#include #include +#include +#include +#include #include +#include namespace livekit::test { @@ -63,4 +68,65 @@ TEST_F(AudioSourceTest, BufferedIngressTimeoutIsNotReportedAsSuccess) { source.clearQueue(); } +TEST_F(AudioSourceTest, FreshSinkBoundsQueueAndPreservesStereo) { + auto source = std::make_shared(48000, 2, 0); + auto track = LocalAudioTrack::createLocalAudioTrack("fresh-input", source); + FreshAudioStream stream(track); + auto input = AudioFrame::create(48000, 2, 480); + for (std::size_t sample = 0; sample < input.data().size(); sample += 2) { + input.data()[sample] = 1234; + input.data()[sample + 1] = -2345; + } + for (int frame = 0; frame < 12; ++frame) source->captureFrame(input, 1000); + FreshAudioFrame output; + ASSERT_EQ(stream.tryRead(output, 1'000'000), FreshAudioReadResult::frame); + EXPECT_EQ(output.samples.front(), 1234); + EXPECT_EQ(output.samples.back(), -2345); + EXPECT_TRUE(output.discontinuity); + EXPECT_LE(stream.stats().queued, 3u); + EXPECT_GE(stream.stats().dropped, 8u); + EXPECT_EQ(stream.tryRead(output, 0), FreshAudioReadResult::invalid_input); + FreshAudioReadResult foreign = FreshAudioReadResult::frame; + std::thread reader([&] { foreign = stream.tryRead(output); }); + reader.join(); + EXPECT_EQ(foreign, FreshAudioReadResult::wrong_thread); +} + +TEST_F(AudioSourceTest, FreshSinkExpiresOldFramesAndConvertsFormat) { + auto source = std::make_shared(16000, 1, 0); + auto track = LocalAudioTrack::createLocalAudioTrack("converted-input", source); + FreshAudioStream stream(track); + auto input = AudioFrame::create(16000, 1, 160); + std::fill(input.data().begin(), input.data().end(), 2000); + for (int frame = 0; frame < 4; ++frame) source->captureFrame(input, 1000); + // An intentional consumer stall must expire queued PCM, not refresh its age + // at read time. This duration is the injected fault, not readiness polling. + std::this_thread::sleep_for(std::chrono::milliseconds(70)); + FreshAudioFrame output; + EXPECT_EQ(stream.tryRead(output, 60'000), FreshAudioReadResult::empty); + EXPECT_GT(stream.stats().stale, 0u); + source->captureFrame(input, 1000); + ASSERT_EQ(stream.tryRead(output, 1'000'000), FreshAudioReadResult::frame); + EXPECT_EQ(output.samples[400], output.samples[401]); + EXPECT_GT(output.samples[400], 1000); + EXPECT_TRUE(output.discontinuity); + EXPECT_EQ(stream.stats().invalid, 0u); +} + +TEST_F(AudioSourceTest, FreshSinkRepeatedTeardownDetachesCallback) { + auto source = std::make_shared(48000, 1, 0); + auto track = LocalAudioTrack::createLocalAudioTrack("lifetime-input", source); + auto input = AudioFrame::create(48000, 1, 480); + for (int cycle = 0; cycle < 50; ++cycle) { + { + FreshAudioStream stream(track); + source->captureFrame(input, 1000); + FreshAudioFrame output; + ASSERT_EQ(stream.tryRead(output, 1'000'000), FreshAudioReadResult::frame); + } + // The retired sink must not remain in the native source's callback list. + source->captureFrame(input, 1000); + } +} + } // namespace livekit::test diff --git a/src/tests/unit/test_room.cpp b/src/tests/unit/test_room.cpp index 2276993..7700267 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -31,6 +31,14 @@ namespace livekit { struct RoomTestAccess { + static void setRemoteParticipant(Room& room, std::shared_ptr participant) { + const std::scoped_lock guard(room.lock_); + room.remote_participants_.clear(); + if (participant) { + room.remote_participants_.emplace(participant->identity(), std::move(participant)); + } + } + static void installConnectedListener(Room& room, std::atomic& callback_count) { const auto listener_id = FfiClient::instance().addListener([&room, &callback_count](const proto::FfiEvent& event) { callback_count.fetch_add(1, std::memory_order_relaxed); @@ -87,6 +95,37 @@ class UnitLifecycleTrackingDelegate : public RoomDelegate { } // namespace +TEST_F(RoomTest, PublicationSnapshotBoundsAndRetainsRemovedHandles) { + Room room; + EXPECT_FALSE(room.remotePublications(0).truncated); + auto participant = std::make_shared(FfiHandle{}, "participant-sid", "name", "identity", "", + std::unordered_map{}, + ParticipantKind::Standard, DisconnectReason::Unknown); + for (std::size_t index = 0; index < 4097; ++index) { + proto::OwnedTrackPublication owned; + const auto sid = "publication-" + std::to_string(index); + owned.mutable_info()->set_sid(sid); + participant->mutableTrackPublications().emplace(sid, std::make_shared(owned)); + } + RoomTestAccess::setRemoteParticipant(room, participant); + EXPECT_TRUE(room.remotePublications(0).truncated); + auto single = room.remotePublications(1); + ASSERT_EQ(single.entries.size(), 1); + EXPECT_TRUE(single.truncated); + const auto capped = room.remotePublications(5000); + EXPECT_EQ(capped.entries.size(), 4096); + EXPECT_TRUE(capped.truncated); + const auto retained_sid = single.entries.front().publication->sid(); + participant->mutableTrackPublications().clear(); + participant->mutableTrackPublications().emplace(retained_sid, single.entries.front().publication); + EXPECT_FALSE(room.remotePublications(1).truncated); + RoomTestAccess::setRemoteParticipant(room, nullptr); + participant.reset(); + EXPECT_TRUE(room.remotePublications(1).entries.empty()); + EXPECT_EQ(single.entries.front().publication->sid(), retained_sid); + EXPECT_EQ(single.entries.front().participant->identity(), "identity"); +} + TEST_F(RoomTest, ConnectWithoutInitialize) { // Test fixture initializes by default, do this to emulate lack of initialization livekit::shutdown();