diff --git a/README.md b/README.md index 459a0ecd..1d6dfb1e 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,23 @@ room->addOnDataFrameCallback(sender_identity, "app-data", }); ``` +Calling `setOnAudioFrameCallback` / `setOnVideoFrameCallback` / +`setOnVideoFrameEventCallback` again for the same +`(participant_identity, track_name)` **replaces** the callback in place. The +previous reader is stopped and its thread joined before the call returns, then a +fresh reader is started bound to the new callback — there is no need to call +`clearOn*FrameCallback` first. Two consequences worth knowing: + +- **These calls block** until any in-flight invocation of the previous callback + returns. When the call returns, the old callback is guaranteed to have + finished and been destroyed. A callback that blocks forever blocks + registration forever. +- **Do not register or clear from inside a frame callback.** Doing so would make + the join a self-join. The SDK detects this, logs an error, and detaches the + reader (media) or leaves it in place to be reaped at teardown (data), but the + registration does not behave as intended. Drive callback changes from another + thread. + For end-to-end samples and a fuller set of demos, see the [cpp-example-collection repo](https://github.com/livekit-examples/cpp-example-collection). ### Generating tokens diff --git a/include/livekit/data_track_stream.h b/include/livekit/data_track_stream.h index f5f07a90..e7e27f10 100644 --- a/include/livekit/data_track_stream.h +++ b/include/livekit/data_track_stream.h @@ -92,9 +92,7 @@ class LIVEKIT_API DataTrackStream { private: friend class RemoteDataTrack; -#ifdef LIVEKIT_TEST_ACCESS friend class DataTrackStreamTest; -#endif DataTrackStream() = default; /// Internal init helper, called by RemoteDataTrack. diff --git a/include/livekit/remote_data_track.h b/include/livekit/remote_data_track.h index 196cde54..60aabfe5 100644 --- a/include/livekit/remote_data_track.h +++ b/include/livekit/remote_data_track.h @@ -92,11 +92,6 @@ class RemoteDataTrack { /// @param options Pipeline options to apply to this remote data track. LIVEKIT_API void setPipelineOptions(const DataTrackPipelineOptions& options); -#ifdef LIVEKIT_TEST_ACCESS - /// Test-only accessor for exercising lower-level FFI subscription paths. - uintptr_t testFfiHandleId() const noexcept { return ffiHandleId(); } -#endif - /// Subscribe to this remote data track. /// /// Returns a DataTrackStream that delivers frames via blocking @@ -106,8 +101,9 @@ class RemoteDataTrack { private: friend class Room; + friend struct RemoteDataTrackTestAccess; - explicit RemoteDataTrack(const proto::OwnedRemoteDataTrack& owned); + LIVEKIT_INTERNAL_API explicit RemoteDataTrack(const proto::OwnedRemoteDataTrack& owned); uintptr_t ffiHandleId() const noexcept { return handle_.get(); } /// RAII wrapper for the Rust-owned FFI resource. diff --git a/include/livekit/room.h b/include/livekit/room.h index faffa715..5404985a 100644 --- a/include/livekit/room.h +++ b/include/livekit/room.h @@ -313,16 +313,15 @@ class LIVEKIT_API Room { // Frame callbacks // --------------------------------------------------------------- - /// @brief Sets the audio frame callback via SubscriptionThreadDispatcher. + /// Register or replace an audio frame callback for a remote subscription via SubscriptionThreadDispatcher. void setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, AudioFrameCallback callback, const AudioStream::Options& opts = {}); - /// @brief Sets the video frame callback via SubscriptionThreadDispatcher. + /// Register or replace a video frame callback for a remote subscription via SubscriptionThreadDispatcher. void setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameCallback callback, const VideoStream::Options& opts = {}); - /// @brief Sets the video frame event callback via - /// SubscriptionThreadDispatcher. + /// Register or replace a video frame event callback for a remote subscription via SubscriptionThreadDispatcher. void setOnVideoFrameEventCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameEventCallback callback, const VideoStream::Options& opts = {}); @@ -364,6 +363,12 @@ class LIVEKIT_API Room { // FfiClient listener ID (0 means no listener registered) int listener_id_{0}; + /// Find a currently subscribed remote track matching the given participant + /// identity and track name. Returns nullptr if no such subscribed track + /// exists. Acquires @ref lock_. + std::shared_ptr findSubscribedRemoteTrack(const std::string& participant_identity, + const std::string& track_name) const; + void onEvent(const proto::FfiEvent& event); // Shared shutdown path for explicit disconnect, server disconnect, EOS, and destruction. diff --git a/include/livekit/subscription_thread_dispatcher.h b/include/livekit/subscription_thread_dispatcher.h index 73c86684..68d228ed 100644 --- a/include/livekit/subscription_thread_dispatcher.h +++ b/include/livekit/subscription_thread_dispatcher.h @@ -16,6 +16,8 @@ #pragma once +#include +#include #include #include #include @@ -65,13 +67,15 @@ using DataFrameCallbackId = std::uint64_t; /// /// `SubscriptionThreadDispatcher` is the low-level companion to @ref Room's /// remote track subscription flow. `Room` forwards user-facing callback -/// registration requests here, and then calls @ref handleTrackSubscribed and -/// @ref handleTrackUnsubscribed as room events arrive. +/// registration requests here. For remote audio and video subscriptions it +/// calls @ref handleTrackSubscribed and @ref handleTrackUnsubscribed; for +/// data tracks it calls @ref handleDataTrackPublished and +/// @ref handleDataTrackUnpublished. /// -/// For each registered `(participant identity, track name)` pair, this class -/// may create a dedicated @ref AudioStream or @ref VideoStream and a matching -/// reader thread. That thread blocks on stream reads and invokes the -/// registered callback with decoded frames. +/// For each registered audio or video `(participant identity, track name)` +/// pair, this class may create a dedicated @ref AudioStream or @ref +/// VideoStream and a matching reader thread. That thread blocks on stream +/// reads and invokes the registered callback with decoded frames. /// /// This type is intentionally independent from @ref RoomDelegate. High-level /// room events such as `RoomDelegate::onTrackSubscribed()` remain in @ref Room, @@ -81,7 +85,7 @@ using DataFrameCallbackId = std::uint64_t; /// The design keeps track-type-specific startup isolated so additional track /// kinds can be added later without pushing more thread state back into /// @ref Room. -class LIVEKIT_API SubscriptionThreadDispatcher { +class LIVEKIT_INTERNAL_API SubscriptionThreadDispatcher { public: /// Constructs an empty dispatcher with no registered callbacks or readers. SubscriptionThreadDispatcher(); @@ -95,6 +99,20 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// If the matching remote audio track is already subscribed, @ref Room may /// immediately call @ref handleTrackSubscribed to start a reader. /// + /// Registering again for a key that already has an active reader replaces the + /// callback in place: the previous reader's stream is closed and its thread + /// is joined before this call returns, and @ref Room then starts a fresh + /// reader bound to the new callback. When this call returns, the previous + /// callback has finished executing and its copy has been destroyed. + /// + /// @warning This call blocks until any in-flight invocation of the previous + /// callback returns. A slow callback makes registration slow; a + /// callback that never returns blocks this call indefinitely. + /// + /// @warning Calling this from inside a frame callback for the same key is not + /// supported. The dispatcher detects the re-entrant call, logs an + /// error, and detaches the reader instead of self-joining. + /// /// @param participant_identity Identity of the remote participant. /// @param track_name Track name to match. /// @param callback Function invoked for each decoded audio frame. @@ -109,6 +127,13 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// If the matching remote video track is already subscribed, @ref Room may /// immediately call @ref handleTrackSubscribed to start a reader. /// + /// Registering again for a key that already has an active reader replaces the + /// callback in place; see @ref setOnAudioFrameCallback for the full + /// replacement semantics, blocking behavior, and re-entrancy caveat. Note + /// that this shares its registration slot with + /// @ref setOnVideoFrameEventCallback -- registering either one replaces the + /// other for the same key. + /// /// @param participant_identity Identity of the remote participant. /// @param track_name Track name to match. /// @param callback Function invoked for each decoded video frame. @@ -124,6 +149,12 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// If the matching remote video track is already subscribed, @ref Room may /// immediately call @ref handleTrackSubscribed to start a reader. /// + /// Registering again for a key that already has an active reader replaces the + /// callback in place; see @ref setOnAudioFrameCallback for the full + /// replacement semantics, blocking behavior, and re-entrancy caveat. Note + /// that this shares its registration slot with @ref setOnVideoFrameCallback + /// -- registering either one replaces the other for the same key. + /// /// @param participant_identity Identity of the remote participant. /// @param track_name Track name to match. /// @param callback Function invoked for each decoded video frame @@ -136,7 +167,13 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Remove an audio callback registration and stop any active reader. /// /// If an audio reader thread is active for the given key, its stream is - /// closed and the thread is joined before this call returns. + /// closed and the thread is joined before this call returns. Replacing a + /// callback does not require clearing first -- see + /// @ref setOnAudioFrameCallback. + /// + /// @warning Blocks until any in-flight callback invocation returns, and is + /// not supported from inside a frame callback for the same key. See + /// @ref setOnAudioFrameCallback. /// /// @param participant_identity Identity of the remote participant. /// @param track_name Track name to clear. @@ -145,33 +182,47 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Remove a video callback registration and stop any active reader. /// /// If a video reader thread is active for the given key, its stream is - /// closed and the thread is joined before this call returns. + /// closed and the thread is joined before this call returns. Replacing a + /// callback does not require clearing first -- see + /// @ref setOnVideoFrameCallback. + /// + /// @warning Blocks until any in-flight callback invocation returns, and is + /// not supported from inside a frame callback for the same key. See + /// @ref setOnAudioFrameCallback. /// /// @param participant_identity Identity of the remote participant. /// @param track_name Track name to clear. void clearOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name); - /// Start or restart reader dispatch for a newly subscribed remote track. + /// Start or restart reader dispatch for a newly subscribed remote audio or + /// video track. /// /// @ref Room calls this after it has processed a track-subscription event and - /// updated its publication state. If a matching callback registration exists, - /// the dispatcher creates the appropriate stream type and launches a reader - /// thread for the `(participant, track_name)` key. + /// updated its publication state. If a matching audio or video callback + /// registration exists, the dispatcher creates the appropriate @ref + /// AudioStream or @ref VideoStream and launches a reader thread for the + /// `(participant, track_name)` key. /// - /// If no matching callback is registered, this is a no-op. + /// Remote data tracks are handled separately via @ref + /// handleDataTrackPublished. If @p track is not audio or video, or no + /// matching callback is registered, this is a no-op. /// /// @param participant_identity Identity of the remote participant. /// @param track_name Track name associated with the subscription. - /// @param track Subscribed remote track to read from. + /// @param track Subscribed remote audio or video track to read + /// from. void handleTrackSubscribed(const std::string& participant_identity, const std::string& track_name, const std::shared_ptr& track); - /// Stop reader dispatch for an unsubscribed remote track. + /// Stop reader dispatch for an unsubscribed remote audio or video track. /// - /// @ref Room calls this when a remote track is unsubscribed. Any active - /// reader stream for the given `(participant, track_name)` key is closed and its - /// thread is joined. Callback registration is preserved so future - /// re-subscription can start dispatch again automatically. + /// @ref Room calls this when a remote audio or video track is unsubscribed. + /// Any active reader stream for the given `(participant, track_name)` key is + /// closed and its thread is joined. Callback registration is preserved so + /// future re-subscription can start dispatch again automatically. + /// + /// Remote data tracks are handled separately via @ref + /// handleDataTrackUnpublished. /// /// @param participant_identity Identity of the remote participant. /// @param source Track source associated with the subscription. @@ -206,6 +257,15 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// for this subscription. /// No-op if the ID is not (or no longer) registered. /// + /// @warning Blocks until any in-flight invocation of the callback returns. + /// + /// @warning Calling this from inside the data frame callback it would remove + /// is not supported. The dispatcher detects the re-entrant call, + /// logs an error, and leaves the reader in place; the reader is + /// reaped on teardown instead. Data readers cannot be safely + /// detached because they re-enter the dispatcher after the callback + /// returns. + /// /// @param id The identifier returned by addOnDataFrameCallback(). void removeOnDataFrameCallback(DataFrameCallbackId id); @@ -234,6 +294,7 @@ class LIVEKIT_API SubscriptionThreadDispatcher { private: friend class SubscriptionThreadDispatcherTest; + friend struct RoomTestAccess; /// Compound lookup key for audio/video callback dispatch. struct CallbackKey { @@ -259,6 +320,13 @@ class LIVEKIT_API SubscriptionThreadDispatcher { std::shared_ptr audio_stream; std::shared_ptr video_stream; std::thread thread; + /// SID of the subscribed track backing this reader, used to skip redundant + /// reader restarts when the same publication is re-subscribed. + std::string track_sid; + /// ID of @ref thread, captured at construction. Used to detect a re-entrant + /// call made from inside this reader's own frame callback, where joining + /// would be a self-join. + std::thread::id thread_id; }; /// Compound lookup key for a remote participant identity and data track name. @@ -289,9 +357,19 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// Active read-side resources for one data track stream subscription. struct ActiveDataReader { std::shared_ptr remote_track; + /// Set true when this reader is being replaced or torn down so the reader + /// thread can abort a subscription that is still in flight. + std::atomic cancelled{false}; + /// Guarded by lock_. Reader threads may mark themselves finished, but only + /// dispatcher lifecycle paths erase the slot and join the thread. + bool finished = false; std::mutex sub_mutex; std::shared_ptr stream; // guarded by sub_mutex std::thread thread; + /// ID of @ref thread, captured at construction. Used to detect a re-entrant + /// call made from inside this reader's own data frame callback, where + /// joining would be a self-join. + std::thread::id thread_id; }; /// Stored audio callback registration plus stream-construction options. @@ -313,7 +391,28 @@ class LIVEKIT_API SubscriptionThreadDispatcher { /// must be joined after releasing the lock. std::thread extractReaderThreadLocked(const CallbackKey& key); - /// Select the appropriate reader startup path for @p track. + /// True when @p id identifies the calling thread, i.e. joining that thread + /// would be a self-join. + static bool isSelfThread(std::thread::id id) { return id == std::this_thread::get_id(); } + + /// Dispose of an extracted audio/video reader thread. + /// + /// Normally joins, so the caller is guaranteed the reader has stopped and its + /// callback copy has been destroyed. If the caller *is* that reader -- a + /// re-entrant registration from inside a frame callback -- joining would be a + /// self-join, so this logs an error naming @p operation and detaches instead. + /// Detaching is safe here because audio/video reader lambdas capture no + /// @c this and own their stream and callback by value. + /// + /// Must be called with @ref lock_ released. + void disposeMediaReaderThread(std::thread&& thread, const char* operation); + + /// Select the appropriate reader startup path for @p media track. + /// + /// This is called by @ref Room when a remote track is subscribed. If a reader + /// for the same track SID is already active, startup is skipped and a + /// default-constructed thread is returned; otherwise any previous reader is + /// extracted and returned to the caller for joining outside the lock. /// /// Must be called with @ref lock_ held. std::thread startReaderLocked(const CallbackKey& key, const std::shared_ptr& track); @@ -333,18 +432,21 @@ class LIVEKIT_API SubscriptionThreadDispatcher { const RegisteredVideoCallback& callback); /// Extract and close the data reader for a given callback ID, returning its - /// thread. Must be called with @ref lock_ held. + /// thread. Marks the reader cancelled so a subscription still in flight is + /// aborted. Must be called with @ref lock_ held. std::thread extractDataReaderThreadLocked(DataFrameCallbackId id); - /// Extract and close the data reader for a given (participant, track_name) - /// key, returning its thread. Must be called with @ref lock_ held. - std::thread extractDataReaderThreadLocked(const DataCallbackKey& key); - /// Start a data reader thread for the given callback ID, key, and track. /// Must be called with @ref lock_ held. std::thread startDataReaderLocked(DataFrameCallbackId id, const DataCallbackKey& key, const std::shared_ptr& track, const DataFrameCallback& cb); + /// Mark @p reader finished if the slot for @p id still refers to it. + /// Called by the reader thread itself when it exits after a failed, + /// cancelled, or terminal subscription. Acquires @ref lock_. Reader threads + /// must not erase, detach, or join their own @ref std::thread. + void markDataReaderFinishedIfCurrent(DataFrameCallbackId id, const std::shared_ptr& reader); + /// Protects callback registration maps and active reader state. mutable std::mutex lock_; diff --git a/src/room.cpp b/src/room.cpp index 8277cf36..ba7d0165 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -400,27 +400,88 @@ void Room::unregisterByteStreamHandler(const std::string& topic) { // Frame callback registration // ------------------------------------------------------------------- +std::shared_ptr Room::findSubscribedRemoteTrack(const std::string& participant_identity, + const std::string& track_name) const { + const std::scoped_lock guard(lock_); + auto pit = remote_participants_.find(participant_identity); + if (pit == remote_participants_.end() || !pit->second) { + return nullptr; + } + for (const auto& [sid, publication] : pit->second->trackPublications()) { + (void)sid; + if (publication && publication->subscribed() && publication->name() == track_name) { + return publication->track(); + } + } + return nullptr; +} + void Room::setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, AudioFrameCallback callback, const AudioStream::Options& opts) { - if (subscription_thread_dispatcher_) { - subscription_thread_dispatcher_->setOnAudioFrameCallback(participant_identity, track_name, std::move(callback), - opts); + if (!subscription_thread_dispatcher_) { + LK_LOG_ERROR("Room::setOnAudioFrameCallback: subscription_thread_dispatcher_ is nullptr"); + return; + } + // Installs the callback and stops any reader still dispatching to the previous + // one, so the restart below binds a fresh reader to the new callback. + subscription_thread_dispatcher_->setOnAudioFrameCallback(participant_identity, track_name, std::move(callback), opts); + + // If we've already subscribed to the track, handle it immediately + auto track = findSubscribedRemoteTrack(participant_identity, track_name); + if (track) { + subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); + } else { + // The track is not subscribed yet. The callback is registered; the reader + // starts when the track is subscribed (see kTrackSubscribed in onEvent). + LK_LOG_DEBUG( + "Room::setOnAudioFrameCallback: track not yet subscribed for participant={} track_name={}; " + "callback registered for deferred start", + participant_identity, track_name); } } void Room::setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameCallback callback, const VideoStream::Options& opts) { - if (subscription_thread_dispatcher_) { - subscription_thread_dispatcher_->setOnVideoFrameCallback(participant_identity, track_name, std::move(callback), - opts); + if (!subscription_thread_dispatcher_) { + LK_LOG_ERROR("Room::setOnVideoFrameCallback: subscription_thread_dispatcher_ is nullptr"); + return; + } + subscription_thread_dispatcher_->setOnVideoFrameCallback(participant_identity, track_name, std::move(callback), opts); + + // If we've already subscribed to the track, handle it immediately + auto track = findSubscribedRemoteTrack(participant_identity, track_name); + if (track) { + subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); + } else { + // The track is not subscribed yet. The callback is registered; the reader + // starts when the track is subscribed (see kTrackSubscribed in onEvent). + LK_LOG_DEBUG( + "Room::setOnVideoFrameCallback: track not yet subscribed for participant={} track_name={}; " + "callback registered for deferred start", + participant_identity, track_name); } } void Room::setOnVideoFrameEventCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameEventCallback callback, const VideoStream::Options& opts) { - if (subscription_thread_dispatcher_) { - subscription_thread_dispatcher_->setOnVideoFrameEventCallback(participant_identity, track_name, std::move(callback), - opts); + if (!subscription_thread_dispatcher_) { + LK_LOG_ERROR("Room::setOnVideoFrameEventCallback: subscription_thread_dispatcher_ is nullptr"); + return; + } + subscription_thread_dispatcher_->setOnVideoFrameEventCallback(participant_identity, track_name, std::move(callback), + opts); + + // If we've already subscribed to the track, handle it immediately + auto track = findSubscribedRemoteTrack(participant_identity, track_name); + if (track) { + subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); + } else { + // The track is not subscribed yet. The callback is registered; the reader + // starts when the track is subscribed (see kTrackSubscribed in onEvent). + LK_LOG_DEBUG( + "Room::setOnVideoFrameEventCallback: track not yet subscribed for participant={} track_name={}; " + "callback registered for deferred start", + participant_identity, track_name); } } diff --git a/src/subscription_thread_dispatcher.cpp b/src/subscription_thread_dispatcher.cpp index ed77d0be..63b88c57 100644 --- a/src/subscription_thread_dispatcher.cpp +++ b/src/subscription_thread_dispatcher.cpp @@ -57,17 +57,46 @@ SubscriptionThreadDispatcher::~SubscriptionThreadDispatcher() { } // NOLINTEND(bugprone-exception-escape) +void SubscriptionThreadDispatcher::disposeMediaReaderThread(std::thread&& thread, const char* operation) { + if (!thread.joinable()) { + return; + } + if (isSelfThread(thread.get_id())) { + // The caller IS this reader, so it called us from inside its own frame + // callback. Joining here would be a self-join. Detaching is safe: audio and + // video reader lambdas capture no `this` and own their stream and callback + // by value, so the thread touches nothing owned by the dispatcher once it + // has been extracted. + LK_LOG_ERROR( + "{} was called from inside its own frame callback; detaching the reader " + "instead of self-joining. Registering or clearing a callback from within " + "that callback is not supported", + operation); + thread.detach(); + return; + } + thread.join(); +} + void SubscriptionThreadDispatcher::setOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, AudioFrameCallback callback, const AudioStream::Options& opts) { const CallbackKey key{participant_identity, track_name}; - const std::scoped_lock lock(lock_); - const bool replacing = audio_callbacks_.find(key) != audio_callbacks_.end(); - audio_callbacks_[key] = RegisteredAudioCallback{std::move(callback), opts}; - LK_LOG_DEBUG( - "Registered audio frame callback for participant={} track_name={} " - "replacing_existing={} total_audio_callbacks={}", - participant_identity, track_name, replacing, audio_callbacks_.size()); + std::thread old_thread; + { + const std::scoped_lock lock(lock_); + // Stop any reader still dispatching to the previous callback. Reader threads + // hold their own copy of the callback, so overwriting the registration alone + // would leave the old callback receiving frames. + old_thread = extractReaderThreadLocked(key); + const bool replacing = audio_callbacks_.find(key) != audio_callbacks_.end(); + audio_callbacks_[key] = RegisteredAudioCallback{std::move(callback), opts}; + LK_LOG_DEBUG( + "Registered audio frame callback for participant={} track_name={} " + "replacing_existing={} stopped_reader={} total_audio_callbacks={}", + participant_identity, track_name, replacing, old_thread.joinable(), audio_callbacks_.size()); + } + disposeMediaReaderThread(std::move(old_thread), "setOnAudioFrameCallback"); } void SubscriptionThreadDispatcher::setOnVideoFrameEventCallback(const std::string& participant_identity, @@ -75,34 +104,44 @@ void SubscriptionThreadDispatcher::setOnVideoFrameEventCallback(const std::strin VideoFrameEventCallback callback, const VideoStream::Options& opts) { const CallbackKey key{participant_identity, track_name}; - const std::scoped_lock lock(lock_); - const bool replacing = video_callbacks_.find(key) != video_callbacks_.end(); - video_callbacks_[key] = RegisteredVideoCallback{ - VideoFrameCallback{}, - std::move(callback), - opts, - }; - LK_LOG_DEBUG( - "Registered video frame event callback for participant={} track_name={} " - "replacing_existing={} total_video_callbacks={}", - participant_identity, track_name, replacing, video_callbacks_.size()); + std::thread old_thread; + { + const std::scoped_lock lock(lock_); + old_thread = extractReaderThreadLocked(key); + const bool replacing = video_callbacks_.find(key) != video_callbacks_.end(); + video_callbacks_[key] = RegisteredVideoCallback{ + VideoFrameCallback{}, + std::move(callback), + opts, + }; + LK_LOG_DEBUG( + "Registered video frame event callback for participant={} track_name={} " + "replacing_existing={} stopped_reader={} total_video_callbacks={}", + participant_identity, track_name, replacing, old_thread.joinable(), video_callbacks_.size()); + } + disposeMediaReaderThread(std::move(old_thread), "setOnVideoFrameEventCallback"); } void SubscriptionThreadDispatcher::setOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, VideoFrameCallback callback, const VideoStream::Options& opts) { const CallbackKey key{participant_identity, track_name}; - const std::scoped_lock lock(lock_); - const bool replacing = video_callbacks_.find(key) != video_callbacks_.end(); - video_callbacks_[key] = RegisteredVideoCallback{ - std::move(callback), - VideoFrameEventCallback{}, - opts, - }; - LK_LOG_DEBUG( - "Registered video frame callback for participant={} track_name={} " - "replacing_existing={} total_video_callbacks={}", - participant_identity, track_name, replacing, video_callbacks_.size()); + std::thread old_thread; + { + const std::scoped_lock lock(lock_); + old_thread = extractReaderThreadLocked(key); + const bool replacing = video_callbacks_.find(key) != video_callbacks_.end(); + video_callbacks_[key] = RegisteredVideoCallback{ + std::move(callback), + VideoFrameEventCallback{}, + opts, + }; + LK_LOG_DEBUG( + "Registered video frame callback for participant={} track_name={} " + "replacing_existing={} stopped_reader={} total_video_callbacks={}", + participant_identity, track_name, replacing, old_thread.joinable(), video_callbacks_.size()); + } + disposeMediaReaderThread(std::move(old_thread), "setOnVideoFrameCallback"); } void SubscriptionThreadDispatcher::clearOnAudioFrameCallback(const std::string& participant_identity, @@ -119,9 +158,7 @@ void SubscriptionThreadDispatcher::clearOnAudioFrameCallback(const std::string& "removed_callback={} stopped_reader={} remaining_audio_callbacks={}", participant_identity, track_name, removed_callback, old_thread.joinable(), audio_callbacks_.size()); } - if (old_thread.joinable()) { - old_thread.join(); - } + disposeMediaReaderThread(std::move(old_thread), "clearOnAudioFrameCallback"); } void SubscriptionThreadDispatcher::clearOnVideoFrameCallback(const std::string& participant_identity, @@ -138,9 +175,7 @@ void SubscriptionThreadDispatcher::clearOnVideoFrameCallback(const std::string& "removed_callback={} stopped_reader={} remaining_video_callbacks={}", participant_identity, track_name, removed_callback, old_thread.joinable(), video_callbacks_.size()); } - if (old_thread.joinable()) { - old_thread.join(); - } + disposeMediaReaderThread(std::move(old_thread), "clearOnVideoFrameCallback"); } void SubscriptionThreadDispatcher::handleTrackSubscribed(const std::string& participant_identity, @@ -161,9 +196,7 @@ void SubscriptionThreadDispatcher::handleTrackSubscribed(const std::string& part const std::scoped_lock lock(lock_); old_thread = startReaderLocked(key, track); } - if (old_thread.joinable()) { - old_thread.join(); - } + disposeMediaReaderThread(std::move(old_thread), "handleTrackSubscribed"); } void SubscriptionThreadDispatcher::handleTrackUnsubscribed(const std::string& participant_identity, TrackSource source, @@ -178,9 +211,7 @@ void SubscriptionThreadDispatcher::handleTrackUnsubscribed(const std::string& pa "track_name={} stopped_reader={}", participant_identity, static_cast(source), track_name, old_thread.joinable()); } - if (old_thread.joinable()) { - old_thread.join(); - } + disposeMediaReaderThread(std::move(old_thread), "handleTrackUnsubscribed"); } // ------------------------------------------------------------------- @@ -259,12 +290,27 @@ void SubscriptionThreadDispatcher::handleDataTrackUnpublished(const std::string& for (auto it = active_data_readers_.begin(); it != active_data_readers_.end();) { auto& reader = it->second; if (reader->remote_track && reader->remote_track->info().sid == sid) { + // Mark cancelled before closing to guard in flight subscriptions + reader->cancelled = true; { const std::scoped_lock sub_guard(reader->sub_mutex); if (reader->stream) { reader->stream->close(); } } + if (isSelfThread(reader->thread_id)) { + // Reached from inside this reader's own data frame callback. It is now + // cancelled and its stream is closed, so it will exit on its own; leave + // the slot for stopAll() to reap rather than self-joining. Data readers + // cannot be detached -- they re-enter the dispatcher on the way out. + LK_LOG_ERROR( + "Data reader for callback id={} reached handleDataTrackUnpublished " + "from inside its own data frame callback; leaving the reader in " + "place to exit on its own", + it->first); + ++it; + continue; + } if (reader->thread.joinable()) { old_threads.push_back(std::move(reader->thread)); } @@ -286,7 +332,10 @@ void SubscriptionThreadDispatcher::handleDataTrackUnpublished(const std::string& } void SubscriptionThreadDispatcher::stopAll() { - std::vector threads; + // Media and data reader threads are disposed of differently: media threads may + // be safely detached on a self-join, data threads may not. + std::vector media_threads; + std::vector data_threads; { const std::scoped_lock lock(lock_); LK_LOG_DEBUG( @@ -304,7 +353,7 @@ void SubscriptionThreadDispatcher::stopAll() { reader.video_stream->close(); } if (reader.thread.joinable()) { - threads.push_back(std::move(reader.thread)); + media_threads.push_back(std::move(reader.thread)); } } active_readers_.clear(); @@ -312,6 +361,8 @@ void SubscriptionThreadDispatcher::stopAll() { video_callbacks_.clear(); for (auto& [id, reader] : active_data_readers_) { + // Mark cancelled before closing to guard in flight subscriptions + reader->cancelled = true; { const std::scoped_lock sub_guard(reader->sub_mutex); if (reader->stream) { @@ -319,17 +370,23 @@ void SubscriptionThreadDispatcher::stopAll() { } } if (reader->thread.joinable()) { - threads.push_back(std::move(reader->thread)); + data_threads.push_back(std::move(reader->thread)); } } active_data_readers_.clear(); data_callbacks_.clear(); remote_data_tracks_.clear(); } - for (auto& thread : threads) { + for (auto& thread : media_threads) { + disposeMediaReaderThread(std::move(thread), "stopAll"); + } + // Data reader threads re-enter the dispatcher after their callback returns, so + // they must be joined even here. Tearing the room down from inside a data + // frame callback is unsupported and will self-join. + for (auto& thread : data_threads) { thread.join(); } - LK_LOG_DEBUG("Stopped {} subscription reader threads", threads.size()); + LK_LOG_DEBUG("Stopped {} subscription reader threads", media_threads.size() + data_threads.size()); } std::thread SubscriptionThreadDispatcher::extractReaderThreadLocked(const CallbackKey& key) { @@ -397,6 +454,16 @@ std::thread SubscriptionThreadDispatcher::startAudioReaderLocked(const CallbackK const AudioFrameCallback& cb, const AudioStream::Options& opts) { LK_LOG_DEBUG("Starting audio reader for participant={} track_name={}", key.participant_identity, key.track_name); + + auto existing = active_readers_.find(key); + if (existing != active_readers_.end() && existing->second.track_sid == track->sid()) { + LK_LOG_DEBUG( + "Skipping audio reader start for participant={} track_name={} because a " + "reader for sid={} is already active", + key.participant_identity, key.track_name, track->sid()); + return {}; + } + auto old_thread = extractReaderThreadLocked(key); if (static_cast(active_readers_.size()) >= kMaxActiveReaders) { @@ -415,6 +482,7 @@ std::thread SubscriptionThreadDispatcher::startAudioReaderLocked(const CallbackK ActiveReader reader; reader.audio_stream = stream; + reader.track_sid = track->sid(); const std::string participant_identity = key.participant_identity; const std::string track_name = key.track_name; // NOLINTBEGIN(bugprone-lambda-function-name,bugprone-exception-escape) @@ -442,6 +510,7 @@ std::thread SubscriptionThreadDispatcher::startAudioReaderLocked(const CallbackK } }); // NOLINTEND(bugprone-lambda-function-name,bugprone-exception-escape) + reader.thread_id = reader.thread.get_id(); active_readers_[key] = std::move(reader); LK_LOG_DEBUG( "Started audio reader for participant={} track_name={} " @@ -454,6 +523,16 @@ std::thread SubscriptionThreadDispatcher::startVideoReaderLocked(const CallbackK const std::shared_ptr& track, const RegisteredVideoCallback& callback) { LK_LOG_DEBUG("Starting video reader for participant={} track_name={}", key.participant_identity, key.track_name); + + auto existing = active_readers_.find(key); + if (existing != active_readers_.end() && existing->second.track_sid == track->sid()) { + LK_LOG_DEBUG( + "Skipping video reader start for participant={} track_name={} because a " + "reader for sid={} is already active", + key.participant_identity, key.track_name, track->sid()); + return {}; + } + auto old_thread = extractReaderThreadLocked(key); if (static_cast(active_readers_.size()) >= kMaxActiveReaders) { @@ -472,6 +551,7 @@ std::thread SubscriptionThreadDispatcher::startVideoReaderLocked(const CallbackK ActiveReader reader; reader.video_stream = stream; + reader.track_sid = track->sid(); auto legacy_cb = callback.legacy_callback; auto event_cb = callback.event_callback; const std::string participant_identity = key.participant_identity; @@ -503,6 +583,7 @@ std::thread SubscriptionThreadDispatcher::startVideoReaderLocked(const CallbackK } }); // NOLINTEND(bugprone-lambda-function-name,bugprone-exception-escape) + reader.thread_id = reader.thread.get_id(); active_readers_[key] = std::move(reader); LK_LOG_DEBUG( "Started video reader for participant={} track_name={} " @@ -520,8 +601,23 @@ std::thread SubscriptionThreadDispatcher::extractDataReaderThreadLocked(DataFram if (it == active_data_readers_.end()) { return {}; } + if (it->second && isSelfThread(it->second->thread_id)) { + // The caller IS this reader, so it reached us from inside its own data frame + // callback. Joining would be a self-join, and unlike media readers a data + // reader cannot be detached: it re-enters the dispatcher after the callback + // returns. Leave the slot in place -- the reader exits on its own once its + // stream closes, and stopAll() reaps it. + LK_LOG_ERROR( + "Data reader for callback id={} tried to tear itself down from inside its " + "own data frame callback; leaving the reader in place. Removing a data " + "callback from within that callback is not supported", + id); + return {}; + } auto reader = std::move(it->second); active_data_readers_.erase(it); + // Mark cancelled before closing to guard in flight subscriptions + reader->cancelled = true; { const std::scoped_lock guard(reader->sub_mutex); if (reader->stream) { @@ -531,28 +627,34 @@ std::thread SubscriptionThreadDispatcher::extractDataReaderThreadLocked(DataFram return std::move(reader->thread); } -std::thread SubscriptionThreadDispatcher::extractDataReaderThreadLocked(const DataCallbackKey& key) { - for (auto it = active_data_readers_.begin(); it != active_data_readers_.end(); ++it) { - if (it->second && it->second->remote_track && - it->second->remote_track->publisherIdentity() == key.participant_identity && - it->second->remote_track->info().name == key.track_name) { - auto reader = std::move(it->second); - active_data_readers_.erase(it); - { - const std::scoped_lock guard(reader->sub_mutex); - if (reader->stream) { - reader->stream->close(); - } - } - return std::move(reader->thread); - } +void SubscriptionThreadDispatcher::markDataReaderFinishedIfCurrent(DataFrameCallbackId id, + const std::shared_ptr& reader) { + const std::scoped_lock lock(lock_); + auto it = active_data_readers_.find(id); + if (it == active_data_readers_.end() || it->second != reader) { + // The slot was already extracted or replaced; the owner joins that thread. + return; + } + reader->finished = true; + { + const std::scoped_lock guard(reader->sub_mutex); + reader->stream.reset(); } - return {}; } std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbackId id, const DataCallbackKey& key, const std::shared_ptr& track, const DataFrameCallback& cb) { + auto existing = active_data_readers_.find(id); + if (existing != active_data_readers_.end() && !existing->second->finished && existing->second->remote_track && + existing->second->remote_track->info().sid == track->info().sid) { + LK_LOG_DEBUG( + "Skipping data reader start for \"{}\" track=\"{}\" because a reader for " + "sid={} is already active", + key.participant_identity, key.track_name, track->info().sid); + return {}; + } + auto old_thread = extractDataReaderThreadLocked(id); const int total_active = static_cast(active_readers_.size()) + static_cast(active_data_readers_.size()); @@ -571,7 +673,7 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac auto identity = key.participant_identity; auto track_name = key.track_name; // NOLINTBEGIN(bugprone-lambda-function-name) - reader->thread = std::thread([reader, track, cb, identity, track_name]() { + reader->thread = std::thread([this, id, reader, track, cb, identity, track_name]() { LK_LOG_INFO("Data reader thread: subscribing to \"{}\" track=\"{}\"", identity, track_name); std::shared_ptr stream; auto subscribe_result = track->subscribe(); @@ -581,14 +683,31 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac "Failed to subscribe to data track \"{}\" from \"{}\": code={} " "message={}", track_name, identity, static_cast(error.code), error.message); + markDataReaderFinishedIfCurrent(id, reader); return; } stream = subscribe_result.value(); LK_LOG_INFO("Data reader thread: subscribed to \"{}\" track=\"{}\"", identity, track_name); + bool cancelled = false; { const std::scoped_lock guard(reader->sub_mutex); - reader->stream = stream; + // A replacement or teardown may have cancelled this reader while the + // subscribe was in flight. Close the fresh stream so we do not leave a + // second live subscription behind. + if (reader->cancelled.load()) { + cancelled = true; + stream->close(); + } else { + reader->stream = stream; + } + } + if (cancelled) { + // Mirror the normal-exit cleanup below. Done outside sub_mutex to keep the + // lock_ -> sub_mutex order and avoid inversion; a no-op unless this reader + // still owns its slot. + markDataReaderFinishedIfCurrent(id, reader); + return; } DataTrackFrame frame; @@ -606,9 +725,13 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac "\"{}\": code={} message={}", track_name, identity, static_cast(error->code), error->message); } + // Mark our own slot finished if the stream ended on its own (server EOS) + // and no extract/teardown already claimed it. A no-op when we were extracted. + markDataReaderFinishedIfCurrent(id, reader); LK_LOG_INFO("Data reader thread exiting for \"{}\" track=\"{}\"", identity, track_name); }); // NOLINTEND(bugprone-lambda-function-name) + reader->thread_id = reader->thread.get_id(); active_data_readers_[id] = reader; return old_thread; } diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt index 9583af73..29a3ece6 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -108,7 +108,6 @@ if(UNIT_TEST_SOURCES) target_compile_definitions(livekit_unit_tests PRIVATE - LIVEKIT_TEST_ACCESS LIVEKIT_ROOT_DIR="${LIVEKIT_ROOT_DIR}" SPDLOG_ACTIVE_LEVEL=${_SPDLOG_ACTIVE_LEVEL} $<$:_USE_MATH_DEFINES> @@ -200,7 +199,6 @@ if(INTEGRATION_TEST_SOURCES) target_compile_definitions(livekit_integration_tests PRIVATE - LIVEKIT_TEST_ACCESS LIVEKIT_ROOT_DIR="${LIVEKIT_ROOT_DIR}" SPDLOG_ACTIVE_LEVEL=${_SPDLOG_ACTIVE_LEVEL} $<$:_USE_MATH_DEFINES> diff --git a/src/tests/common/remote_data_track_test_access.h b/src/tests/common/remote_data_track_test_access.h new file mode 100644 index 00000000..619d1d62 --- /dev/null +++ b/src/tests/common/remote_data_track_test_access.h @@ -0,0 +1,46 @@ +/* + * 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 + +#include "data_track.pb.h" + +namespace livekit { + +struct RemoteDataTrackTestAccess { + static uintptr_t ffiHandleId(const RemoteDataTrack& track) noexcept { return track.ffiHandleId(); } + + static std::shared_ptr create(DataTrackInfo info, std::string publisher_identity) { + proto::OwnedRemoteDataTrack owned; + owned.mutable_handle()->set_id(0); + auto* proto_info = owned.mutable_info(); + proto_info->set_name(std::move(info.name)); + proto_info->set_sid(std::move(info.sid)); + proto_info->set_uses_e2ee(info.uses_e2ee); + owned.set_publisher_identity(std::move(publisher_identity)); + return std::shared_ptr(new RemoteDataTrack(owned)); + } +}; + +} // namespace livekit diff --git a/src/tests/common/room_test_access.h b/src/tests/common/room_test_access.h new file mode 100644 index 00000000..0387b362 --- /dev/null +++ b/src/tests/common/room_test_access.h @@ -0,0 +1,84 @@ +/* + * 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. + */ + +/// @file room_test_access.h +/// @brief In-tree test access to Room internals. +/// +/// Room declares this struct a friend, as does SubscriptionThreadDispatcher, so +/// tests can inspect state that is deliberately not part of the public API. + +#pragma once + +#include +#include + +#include +#include +#include + +#include "ffi.pb.h" +#include "ffi_client.h" + +namespace livekit { + +struct RoomTestAccess { + 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); + room.onEvent(event); + }); + + const std::scoped_lock guard(room.lock_); + room.connection_state_ = ConnectionState::Connected; + room.room_handle_ = std::make_shared(); + room.listener_id_ = listener_id; + } + + static bool hasRoomHandle(const Room& room) { + const std::scoped_lock guard(room.lock_); + return static_cast(room.room_handle_); + } + + static int listenerId(const Room& room) { + const std::scoped_lock guard(room.lock_); + return room.listener_id_; + } + + /// Number of live audio/video reader threads owned by the room's dispatcher. + /// + /// Used by integration tests to prove that replacing a frame callback stops + /// the previous reader rather than leaking one per registration. + static std::size_t activeReaderCount(const Room& room) { + const auto& dispatcher = room.subscription_thread_dispatcher_; + if (!dispatcher) { + return 0; + } + const std::scoped_lock guard(dispatcher->lock_); + return dispatcher->active_readers_.size(); + } + + /// Number of live data track reader threads owned by the room's dispatcher. + static std::size_t activeDataReaderCount(const Room& room) { + const auto& dispatcher = room.subscription_thread_dispatcher_; + if (!dispatcher) { + return 0; + } + const std::scoped_lock guard(dispatcher->lock_); + return dispatcher->active_data_readers_.size(); + } +}; + +} // namespace livekit diff --git a/src/tests/integration/test_data_track.cpp b/src/tests/integration/test_data_track.cpp index d584c551..912b882c 100644 --- a/src/tests/integration/test_data_track.cpp +++ b/src/tests/integration/test_data_track.cpp @@ -25,6 +25,7 @@ #include #include +#include "../common/remote_data_track_test_access.h" #include "../common/test_common.h" #include "ffi_client.h" @@ -465,6 +466,104 @@ TEST_F(DataTrackE2ETest, UnpublishUpdatesPublishedStateEndToEnd) { << "Remote track did not report unpublished state"; } +// Verifies that an auto-wired data callback (Room::addOnDataFrameCallback) +// follows a republished track: after unpublish + republish under the same +// (participant, track name) but a new SID, the previous reader is torn down and +// a fresh reader delivers frames from the new publication. +TEST_F(DataTrackE2ETest, RepublishRewiresDataCallbackToNewPublication) { + const auto track_name = makeTrackName("republish"); + + std::vector room_configs(2); + room_configs[0].room_options.single_peer_connection = false; + room_configs[1].room_options.single_peer_connection = false; + + DataTrackPublishedDelegate subscriber_delegate; + room_configs[1].delegate = &subscriber_delegate; + + auto rooms = testRooms(room_configs); + auto& publisher_room = rooms[0]; + auto& subscriber_room = rooms[1]; + const auto publisher_identity = lockLocalParticipant(*publisher_room)->identity(); + + std::atomic frames_received{0}; + std::mutex payload_mutex; + std::vector last_payload; + subscriber_room->addOnDataFrameCallback(publisher_identity, track_name, + [&](const std::vector& payload, std::optional) { + { + const std::scoped_lock lock(payload_mutex); + last_payload = payload; + } + frames_received.fetch_add(1); + }); + + auto publish_with_retry = [&](const std::string& name) -> std::shared_ptr { + std::shared_ptr track; + waitForCondition( + [&]() { + auto result = lockLocalParticipant(*publisher_room)->publishDataTrack(name); + if (result) { + track = result.value(); + return true; + } + return false; + }, + kTrackWaitTimeout); + return track; + }; + + // First publication. + auto first_track = publish_with_retry(track_name); + ASSERT_NE(first_track, nullptr) << "Failed to publish first data track"; + auto first_remote = subscriber_delegate.waitForTrack(kTrackWaitTimeout); + ASSERT_NE(first_remote, nullptr) << "Timed out waiting for first remote data track"; + const std::string first_sid = first_remote->info().sid; + + DataTrackFrame first_frame; + first_frame.payload.assign(64, 0xA1); + ASSERT_TRUE(waitForCondition( + [&]() { + requirePushSuccess(first_track->tryPush(first_frame), "Failed to push first-publication frame"); + return frames_received.load() > 0; + }, + kTransportFrameTimeout)) + << "Auto-wired callback never received a frame from the first publication"; + + // Unpublish: the reader for the first publication must be torn down. + first_track->unpublishDataTrack(); + ASSERT_TRUE(waitForCondition([&]() { return !first_remote->isPublished(); }, kTrackWaitTimeout)) + << "First remote track did not report unpublished state"; + const int frames_before_republish = frames_received.load(); + + // Republish under the same name; the server assigns a new SID. + auto second_track = publish_with_retry(track_name); + ASSERT_NE(second_track, nullptr) << "Failed to republish data track"; + auto remotes = subscriber_delegate.waitForTracks(2, kTrackWaitTimeout); + ASSERT_EQ(remotes.size(), 2u) << "Timed out waiting for republished remote data track"; + auto second_remote = remotes.back(); + const std::string second_sid = second_remote->info().sid; + EXPECT_NE(first_sid, second_sid) << "Republish should produce a new SID"; + + DataTrackFrame second_frame; + second_frame.payload.assign(64, 0xB2); + ASSERT_TRUE(waitForCondition( + [&]() { + requirePushSuccess(second_track->tryPush(second_frame), "Failed to push republished frame"); + return frames_received.load() > frames_before_republish; + }, + kTransportFrameTimeout)) + << "Auto-wired callback did not re-wire to the republished track"; + + { + const std::scoped_lock lock(payload_mutex); + EXPECT_EQ(last_payload, second_frame.payload) << "Callback delivered stale payload after republish"; + } + + second_track->unpublishDataTrack(); + ASSERT_TRUE(waitForCondition([&]() { return !second_remote->isPublished(); }, kTrackWaitTimeout)) + << "Second remote track did not report unpublished state"; +} + TEST_F(DataTrackE2ETest, SubscribeAfterUnpublishReportsTerminalError) { const auto track_name = makeTrackName("subscribe_after_unpublish"); @@ -854,8 +953,8 @@ TEST_F(DataTrackE2ETest, FfiClientSubscribeDataTrackReturnsSyncResult) { EXPECT_EQ(remote_track->info().name, expected_name); const auto subscribe_start = std::chrono::steady_clock::now(); - auto subscribe_result = - FfiClient::instance().subscribeDataTrack(static_cast(remote_track->testFfiHandleId())); + auto subscribe_result = FfiClient::instance().subscribeDataTrack( + static_cast(RemoteDataTrackTestAccess::ffiHandleId(*remote_track))); const auto subscribe_elapsed = std::chrono::steady_clock::now() - subscribe_start; const auto subscribe_elapsed_ns = std::chrono::duration_cast(subscribe_elapsed).count(); diff --git a/src/tests/integration/test_frame_callback_replacement.cpp b/src/tests/integration/test_frame_callback_replacement.cpp new file mode 100644 index 00000000..ef8774b9 --- /dev/null +++ b/src/tests/integration/test_frame_callback_replacement.cpp @@ -0,0 +1,757 @@ +/* + * 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. + */ + +/// @file test_frame_callback_replacement.cpp +/// @brief End-to-end coverage for in-place frame callback replacement. +/// +/// Registering a frame callback again for the same (participant, track name) +/// replaces it in place: the previous reader is stopped and joined, then a fresh +/// reader is started bound to the new callback. Reader threads hold their own +/// copy of the callback, so without that teardown the old callback keeps +/// receiving frames -- a silent no-op these tests are designed to catch. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "tests/common/audio_utils.h" +#include "tests/common/room_test_access.h" +#include "tests/common/test_common.h" + +namespace livekit::test { + +using namespace std::chrono_literals; + +namespace { + +constexpr auto kSubscribeTimeout = 15s; +constexpr auto kFrameTimeout = 15s; +/// How long to watch a retired callback before concluding it has gone quiet. +constexpr auto kQuiescenceGracePeriod = 1s; +/// Upper bound on any call that must not deadlock. Generous enough to absorb a +/// slow join, tight enough that a real deadlock fails instead of hanging CI. +constexpr auto kNoDeadlockTimeout = 30s; + +constexpr int kFrameWidth = 16; +constexpr int kFrameHeight = 16; + +template +bool waitFor(Predicate predicate, std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (predicate()) { + return true; + } + std::this_thread::sleep_for(10ms); + } + return predicate(); +} + +/// Wait until @p room reports a subscribed track named @p track_name of @p kind +/// published by @p identity. +bool waitForSubscribedTrack(Room& room, const std::string& identity, const std::string& track_name, TrackKind kind, + std::chrono::milliseconds timeout) { + return waitFor( + [&]() { + auto participant = room.remoteParticipant(identity).lock(); + if (participant == nullptr) { + return false; + } + for (const auto& [sid, publication] : participant->trackPublications()) { + (void)sid; + if (publication == nullptr || publication->name() != track_name || publication->kind() != kind) { + continue; + } + if (publication->subscribed() && publication->track() != nullptr) { + return true; + } + } + return false; + }, + timeout); +} + +/// Drives a VideoSource on a background thread for the life of the object. +class VideoPublisher { +public: + explicit VideoPublisher(std::shared_ptr source) : source_(std::move(source)) { + thread_ = std::thread([this]() { + VideoFrame frame = VideoFrame::create(kFrameWidth, kFrameHeight, VideoBufferType::RGBA); + std::fill(frame.data(), frame.data() + frame.dataSize(), 0x7f); + while (running_.load(std::memory_order_relaxed)) { + try { + source_->captureFrame(frame); + } catch (...) { + break; + } + std::this_thread::sleep_for(20ms); + } + }); + } + + VideoPublisher(const VideoPublisher&) = delete; + VideoPublisher& operator=(const VideoPublisher&) = delete; + + void stop() { + running_.store(false, std::memory_order_relaxed); + if (thread_.joinable()) { + thread_.join(); + } + } + + ~VideoPublisher() { stop(); } + +private: + std::shared_ptr source_; + std::atomic running_{true}; + std::thread thread_; +}; + +/// Run @p action on a worker thread and fail if it does not return in time. +/// Used for calls that join a reader thread, where a regression shows up as a +/// hang rather than a wrong value. +[[nodiscard]] bool completesWithoutDeadlock(const std::function& action, + std::chrono::milliseconds timeout = kNoDeadlockTimeout) { + auto future = std::async(std::launch::async, action); + return future.wait_for(timeout) == std::future_status::ready; +} + +/// Two connected rooms plus one published video track, with the receiver +/// confirmed subscribed. Every video test starts from this state. +class VideoFixture { +public: + VideoFixture(const std::string& url, const std::string& token_a, const std::string& token_b, + std::string track_name = "replacement-track") + : track_name_(std::move(track_name)) { + const RoomOptions options; + connected_ = receiver_.connect(url, token_b, options) && sender_.connect(url, token_a, options); + if (!connected_) { + return; + } + if (sender_.localParticipant().expired() || receiver_.localParticipant().expired()) { + connected_ = false; + return; + } + sender_identity_ = lockLocalParticipant(sender_)->identity(); + } + + /// Publish the video track and wait for the receiver to subscribe. + bool publishAndAwaitSubscription() { + if (!connected_ || !waitForParticipant(&receiver_, sender_identity_, kSubscribeTimeout)) { + return false; + } + source_ = std::make_shared(kFrameWidth, kFrameHeight); + track_ = LocalVideoTrack::createLocalVideoTrack(track_name_, source_); + + TrackPublishOptions publish_options; + publish_options.source = TrackSource::SOURCE_CAMERA; + publish_options.simulcast = false; + lockLocalParticipant(sender_)->publishTrack(track_, publish_options); + + publisher_ = std::make_unique(source_); + return waitForSubscribedTrack(receiver_, sender_identity_, track_name_, TrackKind::KIND_VIDEO, kSubscribeTimeout); + } + + void unpublish() { + if (track_ && track_->publication()) { + lockLocalParticipant(sender_)->unpublishTrack(track_->publication()->sid()); + } + } + + void teardown() { + if (publisher_) { + publisher_->stop(); + } + receiver_.clearOnVideoFrameCallback(sender_identity_, track_name_); + unpublish(); + } + + ~VideoFixture() { + if (publisher_) { + publisher_->stop(); + } + } + + bool connected() const { return connected_; } + Room& receiver() { return receiver_; } + Room& sender() { return sender_; } + const std::string& senderIdentity() const { return sender_identity_; } + const std::string& trackName() const { return track_name_; } + const std::shared_ptr& source() const { return source_; } + +private: + Room sender_; + Room receiver_; + std::string track_name_; + std::string sender_identity_; + bool connected_ = false; + std::shared_ptr source_; + std::shared_ptr track_; + std::unique_ptr publisher_; +}; + +/// Assert that @p counter stops advancing, i.e. its callback has been retired. +[[nodiscard]] bool wentQuiet(const std::atomic& counter) { + const int before = counter.load(); + std::this_thread::sleep_for(kQuiescenceGracePeriod); + return counter.load() == before; +} + +} // namespace + +class FrameCallbackReplacementTest : public LiveKitTestBase {}; + +// ============================================================================ +// Core replacement +// ============================================================================ + +// The canonical regression: before the in-place replacement fix, the reader +// thread kept invoking its own copy of callback A forever and B never fired. +TEST_F(FrameCallbackReplacementTest, ReplacingActiveVideoCallbackSwitchesFrameDelivery) { + failIfNotConfigured(); + + VideoFixture fixture(config_.url, config_.token_a, config_.token_b); + ASSERT_TRUE(fixture.connected()); + + std::atomic a_frames{0}; + std::atomic b_frames{0}; + + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [&a_frames](const VideoFrame&, std::int64_t) { a_frames.fetch_add(1); }); + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()); + ASSERT_TRUE(waitFor([&]() { return a_frames.load() > 0; }, kFrameTimeout)) << "First callback never received a frame"; + + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [&b_frames](const VideoFrame&, std::int64_t) { b_frames.fetch_add(1); }); + + EXPECT_TRUE(waitFor([&]() { return b_frames.load() > 0; }, kFrameTimeout)) + << "Replacement callback never received a frame"; + EXPECT_TRUE(wentQuiet(a_frames)) << "Replaced callback is still receiving frames; its reader was not stopped"; + EXPECT_EQ(RoomTestAccess::activeReaderCount(fixture.receiver()), 1u); + + fixture.teardown(); +} + +TEST_F(FrameCallbackReplacementTest, ReplacingActiveAudioCallbackSwitchesFrameDelivery) { + failIfNotConfigured(); + + Room sender_room; + Room receiver_room; + const RoomOptions options; + ASSERT_TRUE(receiver_room.connect(config_.url, config_.token_b, options)); + ASSERT_TRUE(sender_room.connect(config_.url, config_.token_a, options)); + + const std::string sender_identity = lockLocalParticipant(sender_room)->identity(); + ASSERT_TRUE(waitForParticipant(&receiver_room, sender_identity, kSubscribeTimeout)); + + const std::string track_name = "replacement-audio"; + std::atomic a_frames{0}; + std::atomic b_frames{0}; + + receiver_room.setOnAudioFrameCallback(sender_identity, track_name, [&a_frames](const AudioFrame& frame) { + if (frame.totalSamples() > 0) { + a_frames.fetch_add(1); + } + }); + + auto source = std::make_shared(kDefaultAudioSampleRate, kDefaultAudioChannels); + auto track = LocalAudioTrack::createLocalAudioTrack(track_name, source); + TrackPublishOptions publish_options; + publish_options.source = TrackSource::SOURCE_MICROPHONE; + lockLocalParticipant(sender_room)->publishTrack(track, publish_options); + + std::atomic publishing{true}; + std::thread publisher([&]() { runToneLoop(source, publishing, 440.0, /*siren_mode=*/false); }); + + ASSERT_TRUE( + waitForSubscribedTrack(receiver_room, sender_identity, track_name, TrackKind::KIND_AUDIO, kSubscribeTimeout)); + ASSERT_TRUE(waitFor([&]() { return a_frames.load() > 0; }, kFrameTimeout)) << "First callback never received a frame"; + + receiver_room.setOnAudioFrameCallback(sender_identity, track_name, [&b_frames](const AudioFrame& frame) { + if (frame.totalSamples() > 0) { + b_frames.fetch_add(1); + } + }); + + EXPECT_TRUE(waitFor([&]() { return b_frames.load() > 0; }, kFrameTimeout)) + << "Replacement callback never received a frame"; + EXPECT_TRUE(wentQuiet(a_frames)) << "Replaced callback is still receiving frames"; + EXPECT_EQ(RoomTestAccess::activeReaderCount(receiver_room), 1u); + + publishing.store(false); + publisher.join(); + receiver_room.clearOnAudioFrameCallback(sender_identity, track_name); + if (track->publication()) { + lockLocalParticipant(sender_room)->unpublishTrack(track->publication()->sid()); + } +} + +// The legacy and event video callbacks share one registration slot, so +// registering either must displace the other and stop its reader. +TEST_F(FrameCallbackReplacementTest, ReplacingVideoCallbackWithEventCallbackSwitchesDelivery) { + failIfNotConfigured(); + + VideoFixture fixture(config_.url, config_.token_a, config_.token_b); + ASSERT_TRUE(fixture.connected()); + + std::atomic legacy_frames{0}; + std::atomic event_frames{0}; + + fixture.receiver().setOnVideoFrameCallback( + fixture.senderIdentity(), fixture.trackName(), + [&legacy_frames](const VideoFrame&, std::int64_t) { legacy_frames.fetch_add(1); }); + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()); + ASSERT_TRUE(waitFor([&]() { return legacy_frames.load() > 0; }, kFrameTimeout)); + + fixture.receiver().setOnVideoFrameEventCallback( + fixture.senderIdentity(), fixture.trackName(), + [&event_frames](const VideoFrameEvent&) { event_frames.fetch_add(1); }); + + EXPECT_TRUE(waitFor([&]() { return event_frames.load() > 0; }, kFrameTimeout)) + << "Event callback never received a frame after displacing the legacy callback"; + EXPECT_TRUE(wentQuiet(legacy_frames)) << "Displaced legacy callback is still receiving frames"; + EXPECT_EQ(RoomTestAccess::activeReaderCount(fixture.receiver()), 1u); + + fixture.teardown(); +} + +// ============================================================================ +// Long-running callbacks +// ============================================================================ + +// Replacement joins the previous reader, so it must wait out an in-flight +// callback invocation rather than abandoning it mid-frame. This is the guarantee +// the API documents: when the setter returns, the old callback has finished. +TEST_F(FrameCallbackReplacementTest, SetOnVideoFrameCallbackBlocksUntilSlowCallbackReturns) { + failIfNotConfigured(); + + VideoFixture fixture(config_.url, config_.token_a, config_.token_b); + ASSERT_TRUE(fixture.connected()); + + constexpr auto kSlowCallbackDuration = 2s; + std::atomic slow_entered{false}; + std::atomic slow_exited{false}; + std::atomic fast_frames{0}; + + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [&](const VideoFrame&, std::int64_t) { + slow_entered.store(true); + std::this_thread::sleep_for(kSlowCallbackDuration); + slow_exited.store(true); + }); + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()); + ASSERT_TRUE(waitFor([&]() { return slow_entered.load(); }, kFrameTimeout)) + << "Slow callback never started an invocation"; + ASSERT_FALSE(slow_exited.load()) << "Slow callback finished before the replacement was attempted"; + + const auto started_at = std::chrono::steady_clock::now(); + bool exited_before_return = false; + const bool completed = completesWithoutDeadlock([&]() { + fixture.receiver().setOnVideoFrameCallback( + fixture.senderIdentity(), fixture.trackName(), + [&fast_frames](const VideoFrame&, std::int64_t) { fast_frames.fetch_add(1); }); + exited_before_return = slow_exited.load(); + }); + const auto elapsed = std::chrono::steady_clock::now() - started_at; + + ASSERT_TRUE(completed) << "setOnVideoFrameCallback did not return; the join deadlocked"; + EXPECT_TRUE(exited_before_return) << "Setter returned while the previous callback was still executing"; + EXPECT_GE(elapsed, 500ms) << "Setter returned too quickly to have waited on the in-flight callback"; + + EXPECT_TRUE(waitFor([&]() { return fast_frames.load() > 0; }, kFrameTimeout)); + EXPECT_EQ(RoomTestAccess::activeReaderCount(fixture.receiver()), 1u); + + fixture.teardown(); +} + +// The join happens outside the dispatcher lock, so a slow callback on one +// subscription must not stall readers for other subscriptions. +TEST_F(FrameCallbackReplacementTest, SlowCallbackDoesNotStallOtherSubscriptionReaders) { + failIfNotConfigured(); + + Room sender_room; + Room receiver_room; + const RoomOptions options; + ASSERT_TRUE(receiver_room.connect(config_.url, config_.token_b, options)); + ASSERT_TRUE(sender_room.connect(config_.url, config_.token_a, options)); + + const std::string sender_identity = lockLocalParticipant(sender_room)->identity(); + ASSERT_TRUE(waitForParticipant(&receiver_room, sender_identity, kSubscribeTimeout)); + + const std::string slow_track_name = "slow-track"; + const std::string fast_track_name = "fast-track"; + + std::atomic slow_invocations{0}; + std::atomic fast_frames{0}; + + receiver_room.setOnVideoFrameCallback(sender_identity, slow_track_name, [&](const VideoFrame&, std::int64_t) { + slow_invocations.fetch_add(1); + std::this_thread::sleep_for(2s); + }); + receiver_room.setOnVideoFrameCallback(sender_identity, fast_track_name, + [&fast_frames](const VideoFrame&, std::int64_t) { fast_frames.fetch_add(1); }); + + auto slow_source = std::make_shared(kFrameWidth, kFrameHeight); + auto fast_source = std::make_shared(kFrameWidth, kFrameHeight); + auto slow_track = LocalVideoTrack::createLocalVideoTrack(slow_track_name, slow_source); + auto fast_track = LocalVideoTrack::createLocalVideoTrack(fast_track_name, fast_source); + + TrackPublishOptions publish_options; + publish_options.source = TrackSource::SOURCE_CAMERA; + publish_options.simulcast = false; + lockLocalParticipant(sender_room)->publishTrack(slow_track, publish_options); + lockLocalParticipant(sender_room)->publishTrack(fast_track, publish_options); + + VideoPublisher slow_publisher(slow_source); + VideoPublisher fast_publisher(fast_source); + + ASSERT_TRUE(waitForSubscribedTrack(receiver_room, sender_identity, slow_track_name, TrackKind::KIND_VIDEO, + kSubscribeTimeout)); + ASSERT_TRUE(waitForSubscribedTrack(receiver_room, sender_identity, fast_track_name, TrackKind::KIND_VIDEO, + kSubscribeTimeout)); + ASSERT_TRUE(waitFor([&]() { return slow_invocations.load() > 0; }, kFrameTimeout)); + + const int fast_before = fast_frames.load(); + const bool completed = completesWithoutDeadlock([&]() { + receiver_room.setOnVideoFrameCallback(sender_identity, slow_track_name, [](const VideoFrame&, std::int64_t) {}); + }); + ASSERT_TRUE(completed) << "Replacing the slow callback deadlocked"; + + EXPECT_GT(fast_frames.load(), fast_before) + << "The unrelated fast reader stalled while the slow callback was being replaced"; + EXPECT_EQ(RoomTestAccess::activeReaderCount(receiver_room), 2u); + + slow_publisher.stop(); + fast_publisher.stop(); + receiver_room.clearOnVideoFrameCallback(sender_identity, slow_track_name); + receiver_room.clearOnVideoFrameCallback(sender_identity, fast_track_name); + if (slow_track->publication()) { + lockLocalParticipant(sender_room)->unpublishTrack(slow_track->publication()->sid()); + } + if (fast_track->publication()) { + lockLocalParticipant(sender_room)->unpublishTrack(fast_track->publication()->sid()); + } +} + +// ============================================================================ +// Multiple and concurrent calls +// ============================================================================ + +// Each replacement must stop exactly one reader and start exactly one. A leak +// shows up as a growing reader count or as several generations firing at once. +TEST_F(FrameCallbackReplacementTest, RepeatedReplacementUnderLoadLeavesExactlyOneActiveReader) { + failIfNotConfigured(); + + constexpr int kGenerations = 20; + + VideoFixture fixture(config_.url, config_.token_a, config_.token_b); + ASSERT_TRUE(fixture.connected()); + + std::vector>> counters; + counters.reserve(kGenerations); + for (int i = 0; i < kGenerations; ++i) { + counters.push_back(std::make_unique>(0)); + } + + auto* first = counters.front().get(); + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [first](const VideoFrame&, std::int64_t) { first->fetch_add(1); }); + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()); + ASSERT_TRUE(waitFor([&]() { return first->load() > 0; }, kFrameTimeout)); + + for (int i = 1; i < kGenerations; ++i) { + auto* counter = counters[static_cast(i)].get(); + const bool completed = completesWithoutDeadlock([&]() { + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [counter](const VideoFrame&, std::int64_t) { counter->fetch_add(1); }); + }); + ASSERT_TRUE(completed) << "Replacement " << i << " deadlocked"; + ASSERT_LE(RoomTestAccess::activeReaderCount(fixture.receiver()), 1u) + << "Reader count grew during replacement " << i; + std::this_thread::sleep_for(50ms); + } + + auto* last = counters.back().get(); + EXPECT_TRUE(waitFor([&]() { return last->load() > 0; }, kFrameTimeout)) + << "The final callback generation never received a frame"; + EXPECT_EQ(RoomTestAccess::activeReaderCount(fixture.receiver()), 1u); + + // Every earlier generation must be retired: snapshot all of them, wait, and + // confirm none advanced. + std::vector before; + before.reserve(counters.size()); + for (const auto& counter : counters) { + before.push_back(counter->load()); + } + std::this_thread::sleep_for(kQuiescenceGracePeriod); + for (std::size_t i = 0; i + 1 < counters.size(); ++i) { + EXPECT_EQ(counters[i]->load(), before[i]) << "Retired callback generation " << i << " is still receiving frames"; + } + + fixture.teardown(); +} + +// Concurrent replacements must serialize on the dispatcher lock without +// deadlocking, double-starting readers, or losing the final registration. +TEST_F(FrameCallbackReplacementTest, ConcurrentReplacementFromMultipleThreadsIsSerialized) { + failIfNotConfigured(); + + constexpr int kThreads = 4; + constexpr auto kChurnDuration = 2s; + + VideoFixture fixture(config_.url, config_.token_a, config_.token_b); + ASSERT_TRUE(fixture.connected()); + + std::atomic initial_frames{0}; + fixture.receiver().setOnVideoFrameCallback( + fixture.senderIdentity(), fixture.trackName(), + [&initial_frames](const VideoFrame&, std::int64_t) { initial_frames.fetch_add(1); }); + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()); + ASSERT_TRUE(waitFor([&]() { return initial_frames.load() > 0; }, kFrameTimeout)); + + std::atomic churning{true}; + std::atomic max_readers_seen{0}; + std::vector workers; + workers.reserve(kThreads); + for (int i = 0; i < kThreads; ++i) { + workers.emplace_back([&]() { + while (churning.load(std::memory_order_relaxed)) { + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [](const VideoFrame&, std::int64_t) {}); + const auto readers = static_cast(RoomTestAccess::activeReaderCount(fixture.receiver())); + int previous = max_readers_seen.load(); + while (readers > previous && !max_readers_seen.compare_exchange_weak(previous, readers)) { + } + std::this_thread::sleep_for(10ms); + } + }); + } + + auto churn_done = std::async(std::launch::async, [&]() { + std::this_thread::sleep_for(kChurnDuration); + churning.store(false, std::memory_order_relaxed); + for (auto& worker : workers) { + worker.join(); + } + }); + ASSERT_EQ(churn_done.wait_for(kNoDeadlockTimeout), std::future_status::ready) << "Concurrent replacement deadlocked"; + + EXPECT_LE(max_readers_seen.load(), 1) << "Concurrent replacements started more than one reader for the same key"; + + // The registration surviving the churn must still deliver frames. + std::atomic final_frames{0}; + fixture.receiver().setOnVideoFrameCallback( + fixture.senderIdentity(), fixture.trackName(), + [&final_frames](const VideoFrame&, std::int64_t) { final_frames.fetch_add(1); }); + EXPECT_TRUE(waitFor([&]() { return final_frames.load() > 0; }, kFrameTimeout)) + << "Frame delivery did not recover after concurrent replacement"; + EXPECT_EQ(RoomTestAccess::activeReaderCount(fixture.receiver()), 1u); + + fixture.teardown(); +} + +// Deferred start: with no subscription yet there is no reader to stop, and the +// newest registration is the one the reader must bind when the track arrives. +TEST_F(FrameCallbackReplacementTest, ReplacingCallbackBeforeSubscriptionUsesNewestCallback) { + failIfNotConfigured(); + + VideoFixture fixture(config_.url, config_.token_a, config_.token_b); + ASSERT_TRUE(fixture.connected()); + + std::atomic a_frames{0}; + std::atomic b_frames{0}; + + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [&a_frames](const VideoFrame&, std::int64_t) { a_frames.fetch_add(1); }); + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [&b_frames](const VideoFrame&, std::int64_t) { b_frames.fetch_add(1); }); + ASSERT_EQ(RoomTestAccess::activeReaderCount(fixture.receiver()), 0u) << "No reader should exist before subscription"; + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()); + + EXPECT_TRUE(waitFor([&]() { return b_frames.load() > 0; }, kFrameTimeout)) + << "The newest pre-subscription callback never received a frame"; + EXPECT_EQ(a_frames.load(), 0) << "The overwritten pre-subscription callback must never fire"; + EXPECT_EQ(RoomTestAccess::activeReaderCount(fixture.receiver()), 1u); + + fixture.teardown(); +} + +// Replacement while unsubscribed must survive the resubscribe: the registration +// persists across unpublish, and the new callback binds on republish. +TEST_F(FrameCallbackReplacementTest, ReplacementSurvivesUnpublishAndRepublish) { + failIfNotConfigured(); + + VideoFixture fixture(config_.url, config_.token_a, config_.token_b); + ASSERT_TRUE(fixture.connected()); + + std::atomic a_frames{0}; + std::atomic b_frames{0}; + + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [&a_frames](const VideoFrame&, std::int64_t) { a_frames.fetch_add(1); }); + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()); + ASSERT_TRUE(waitFor([&]() { return a_frames.load() > 0; }, kFrameTimeout)); + + fixture.unpublish(); + ASSERT_TRUE(waitFor([&]() { return RoomTestAccess::activeReaderCount(fixture.receiver()) == 0u; }, kSubscribeTimeout)) + << "Reader was not torn down on unpublish"; + + fixture.receiver().setOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName(), + [&b_frames](const VideoFrame&, std::int64_t) { b_frames.fetch_add(1); }); + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()) << "Republished track was never subscribed"; + EXPECT_TRUE(waitFor([&]() { return b_frames.load() > 0; }, kFrameTimeout)) + << "Replacement callback never received a frame from the new publication"; + EXPECT_TRUE(wentQuiet(a_frames)) << "Replaced callback is still receiving frames after republish"; + EXPECT_EQ(RoomTestAccess::activeReaderCount(fixture.receiver()), 1u); + + fixture.teardown(); +} + +// ============================================================================ +// Re-entrancy +// ============================================================================ + +// Registering from inside the frame callback would make the join a self-join. +// Media readers are detached instead, so the call must return rather than hang. +TEST_F(FrameCallbackReplacementTest, SetOnVideoFrameCallbackFromInsideCallbackDoesNotDeadlock) { + failIfNotConfigured(); + + VideoFixture fixture(config_.url, config_.token_a, config_.token_b); + ASSERT_TRUE(fixture.connected()); + + std::atomic reentrant_call_returned{false}; + std::atomic attempted{false}; + std::atomic replacement_frames{0}; + + fixture.receiver().setOnVideoFrameCallback( + fixture.senderIdentity(), fixture.trackName(), [&](const VideoFrame&, std::int64_t) { + if (attempted.exchange(true)) { + return; + } + fixture.receiver().setOnVideoFrameCallback( + fixture.senderIdentity(), fixture.trackName(), + [&replacement_frames](const VideoFrame&, std::int64_t) { replacement_frames.fetch_add(1); }); + reentrant_call_returned.store(true); + }); + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()); + EXPECT_TRUE(waitFor([&]() { return reentrant_call_returned.load(); }, kNoDeadlockTimeout)) + << "Re-entrant setOnVideoFrameCallback never returned; the reader self-joined"; + + // The detached reader exits on its own, and teardown must still complete. + const bool torn_down = completesWithoutDeadlock([&]() { fixture.teardown(); }); + EXPECT_TRUE(torn_down) << "Teardown deadlocked after a re-entrant registration"; +} + +TEST_F(FrameCallbackReplacementTest, ClearOnVideoFrameCallbackFromInsideCallbackDoesNotDeadlock) { + failIfNotConfigured(); + + VideoFixture fixture(config_.url, config_.token_a, config_.token_b); + ASSERT_TRUE(fixture.connected()); + + std::atomic reentrant_call_returned{false}; + std::atomic attempted{false}; + + fixture.receiver().setOnVideoFrameCallback( + fixture.senderIdentity(), fixture.trackName(), [&](const VideoFrame&, std::int64_t) { + if (attempted.exchange(true)) { + return; + } + fixture.receiver().clearOnVideoFrameCallback(fixture.senderIdentity(), fixture.trackName()); + reentrant_call_returned.store(true); + }); + + ASSERT_TRUE(fixture.publishAndAwaitSubscription()); + EXPECT_TRUE(waitFor([&]() { return reentrant_call_returned.load(); }, kNoDeadlockTimeout)) + << "Re-entrant clearOnVideoFrameCallback never returned; the reader self-joined"; + + const bool torn_down = completesWithoutDeadlock([&]() { fixture.teardown(); }); + EXPECT_TRUE(torn_down) << "Teardown deadlocked after a re-entrant clear"; +} + +// Data readers re-enter the dispatcher after their callback returns, so they +// cannot be detached. The re-entrant removal is refused and the reader is left +// for teardown to reap -- which must still join cleanly. +TEST_F(FrameCallbackReplacementTest, RemoveDataCallbackFromInsideDataCallbackIsRefusedWithoutDeadlock) { + failIfNotConfigured(); + + Room sender_room; + Room receiver_room; + const RoomOptions options; + ASSERT_TRUE(receiver_room.connect(config_.url, config_.token_b, options)); + ASSERT_TRUE(sender_room.connect(config_.url, config_.token_a, options)); + + const std::string sender_identity = lockLocalParticipant(sender_room)->identity(); + ASSERT_TRUE(waitForParticipant(&receiver_room, sender_identity, kSubscribeTimeout)); + + const std::string track_name = "reentrant-data"; + std::atomic reentrant_call_returned{false}; + std::atomic attempted{false}; + DataFrameCallbackId callback_id = 0; + + callback_id = receiver_room.addOnDataFrameCallback( + sender_identity, track_name, [&](const std::vector&, std::optional) { + if (attempted.exchange(true)) { + return; + } + receiver_room.removeOnDataFrameCallback(callback_id); + reentrant_call_returned.store(true); + }); + + auto publish_result = lockLocalParticipant(sender_room)->publishDataTrack(track_name); + ASSERT_TRUE(publish_result) << "Failed to publish data track"; + auto local_track = publish_result.value(); + + std::atomic pushing{true}; + std::thread pusher([&]() { + DataTrackFrame frame; + frame.payload.assign(32, 0x5A); + while (pushing.load(std::memory_order_relaxed)) { + (void)local_track->tryPush(frame); + std::this_thread::sleep_for(50ms); + } + }); + + EXPECT_TRUE(waitFor([&]() { return reentrant_call_returned.load(); }, kNoDeadlockTimeout)) + << "Re-entrant removeOnDataFrameCallback never returned; the data reader self-joined"; + + pushing.store(false, std::memory_order_relaxed); + pusher.join(); + + // The refused removal left the reader in place; disconnect must still reap it. + const bool disconnected = completesWithoutDeadlock([&]() { + local_track->unpublishDataTrack(); + receiver_room.disconnect(); + }); + EXPECT_TRUE(disconnected) << "Disconnect deadlocked while reaping the refused data reader"; +} + +} // namespace livekit::test diff --git a/src/tests/integration/test_room_event_deduplication.cpp b/src/tests/integration/test_room_event_deduplication.cpp new file mode 100644 index 00000000..a355afa7 --- /dev/null +++ b/src/tests/integration/test_room_event_deduplication.cpp @@ -0,0 +1,408 @@ +/* + * 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 +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "../common/audio_utils.h" +#include "../common/test_common.h" +#include "../common/video_utils.h" + +namespace livekit::test { + +using namespace std::chrono_literals; + +namespace { + +constexpr auto kEventWaitTimeout = 20s; +constexpr auto kDuplicateGracePeriod = 500ms; + +struct RoomEventCounts { + std::mutex mutex; + std::condition_variable cv; + std::map participant_connected; + std::map participant_disconnected; + std::map track_published; + std::map track_subscribed; + std::map track_unsubscribed; + std::map track_unpublished; + int disconnected = 0; +}; + +struct RoomEventCountsSnapshot { + std::map participant_connected; + std::map participant_disconnected; + std::map track_published; + std::map track_subscribed; + std::map track_unsubscribed; + std::map track_unpublished; + int disconnected = 0; +}; + +RoomEventCountsSnapshot snapshotCounts(RoomEventCounts& counts) { + const std::scoped_lock lock(counts.mutex); + RoomEventCountsSnapshot snapshot; + snapshot.participant_connected = counts.participant_connected; + snapshot.participant_disconnected = counts.participant_disconnected; + snapshot.track_published = counts.track_published; + snapshot.track_subscribed = counts.track_subscribed; + snapshot.track_unsubscribed = counts.track_unsubscribed; + snapshot.track_unpublished = counts.track_unpublished; + snapshot.disconnected = counts.disconnected; + return snapshot; +} + +void incrementMap(std::map& counts, const std::string& key) { ++counts[key]; } + +class RoomEventCounterDelegate : public RoomDelegate { +public: + explicit RoomEventCounterDelegate(RoomEventCounts& counts) : counts_(counts) {} + + void onParticipantConnected(Room&, const ParticipantConnectedEvent& event) override { + if (event.participant == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.participant_connected, event.participant->identity()); }); + } + + void onParticipantDisconnected(Room&, const ParticipantDisconnectedEvent& event) override { + if (event.participant == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.participant_disconnected, event.participant->identity()); }); + } + + void onTrackPublished(Room&, const TrackPublishedEvent& event) override { + if (event.publication == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.track_published, event.publication->name()); }); + } + + void onTrackSubscribed(Room&, const TrackSubscribedEvent& event) override { + if (event.publication == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.track_subscribed, event.publication->name()); }); + } + + void onTrackUnsubscribed(Room&, const TrackUnsubscribedEvent& event) override { + if (event.publication == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.track_unsubscribed, event.publication->name()); }); + } + + void onTrackUnpublished(Room&, const TrackUnpublishedEvent& event) override { + if (event.publication == nullptr) { + return; + } + notify([&]() { incrementMap(counts_.track_unpublished, event.publication->name()); }); + } + + void onDisconnected(Room&, const DisconnectedEvent&) override { + notify([&]() { ++counts_.disconnected; }); + } + +private: + template + void notify(Fn&& update) { + { + const std::scoped_lock lock(counts_.mutex); + update(); + } + counts_.cv.notify_all(); + } + + RoomEventCounts& counts_; +}; + +bool waitForMapCountAtLeast(RoomEventCounts& counts, const std::map& keys, int minimum_count, + std::chrono::milliseconds timeout) { + std::unique_lock lock(counts.mutex); + return counts.cv.wait_for(lock, timeout, [&]() { + for (const auto& [key, _] : keys) { + (void)_; + const auto it = counts.participant_connected.find(key); + if (it == counts.participant_connected.end() || it->second < minimum_count) { + return false; + } + } + return true; + }); +} + +bool waitForMapCountAtLeastTrack(RoomEventCounts& counts, const std::map& expected, + std::map RoomEventCounts::* member, + std::chrono::milliseconds timeout) { + std::unique_lock lock(counts.mutex); + return counts.cv.wait_for(lock, timeout, [&]() { + const auto& actual = counts.*member; + for (const auto& [key, minimum_count] : expected) { + const auto it = actual.find(key); + if (it == actual.end() || it->second < minimum_count) { + return false; + } + } + return true; + }); +} + +void expectMapCountsExact(const std::map& actual, const std::map& expected, + const char* label) { + for (const auto& [key, expected_count] : expected) { + const auto it = actual.find(key); + const int actual_count = it == actual.end() ? 0 : it->second; + EXPECT_EQ(actual_count, expected_count) << label << " count mismatch for key: " << key; + } +} + +void expectCountsUnchangedAfterGrace(RoomEventCounts& counts, const RoomEventCountsSnapshot& before, + const char* phase) { + std::this_thread::sleep_for(kDuplicateGracePeriod); + const RoomEventCountsSnapshot after = snapshotCounts(counts); + + expectMapCountsExact(after.participant_connected, before.participant_connected, + (std::string(phase) + " participant_connected duplicate").c_str()); + expectMapCountsExact(after.participant_disconnected, before.participant_disconnected, + (std::string(phase) + " participant_disconnected duplicate").c_str()); + expectMapCountsExact(after.track_published, before.track_published, + (std::string(phase) + " track_published duplicate").c_str()); + expectMapCountsExact(after.track_subscribed, before.track_subscribed, + (std::string(phase) + " track_subscribed duplicate").c_str()); + expectMapCountsExact(after.track_unsubscribed, before.track_unsubscribed, + (std::string(phase) + " track_unsubscribed duplicate").c_str()); + expectMapCountsExact(after.track_unpublished, before.track_unpublished, + (std::string(phase) + " track_unpublished duplicate").c_str()); + EXPECT_EQ(after.disconnected, before.disconnected) << phase << " onDisconnected duplicate"; +} + +std::string makeUniqueTrackName(const std::string& prefix) { return prefix + "-" + std::to_string(getTimestampUs()); } + +std::string describeCounts(const std::map& counts) { + std::ostringstream out; + bool first = true; + out << "{"; + for (const auto& [key, count] : counts) { + if (!first) { + out << ", "; + } + first = false; + out << key << ": " << count; + } + out << "}"; + return out.str(); +} + +class MediaLoopGuard { +public: + MediaLoopGuard() = default; + MediaLoopGuard(const MediaLoopGuard&) = delete; + MediaLoopGuard& operator=(const MediaLoopGuard&) = delete; + + ~MediaLoopGuard() { stop(); } + + void addAudioSource(const std::shared_ptr& source) { + threads_.emplace_back([this, source]() { + runToneLoop(source, running_, 440.0, false, kDefaultAudioSampleRate, kDefaultAudioChannels); + }); + } + + void addVideoSource(const std::shared_ptr& source) { + threads_.emplace_back([this, source]() { runVideoLoop(source, running_, fillWebcamWrapper); }); + } + + void stop() { + running_.store(false, std::memory_order_relaxed); + for (auto& thread : threads_) { + if (thread.joinable()) { + thread.join(); + } + } + } + +private: + std::atomic running_{true}; + std::vector threads_; +}; + +class PublishedTrackGuard { +public: + explicit PublishedTrackGuard(LocalParticipant* participant) : participant_(participant) {} + PublishedTrackGuard(const PublishedTrackGuard&) = delete; + PublishedTrackGuard& operator=(const PublishedTrackGuard&) = delete; + + ~PublishedTrackGuard() noexcept { + try { + unpublishAll(); + } catch (...) { + } + } + + void addTrackSid(const std::string& sid) { + if (!sid.empty()) { + track_sids_.push_back(sid); + } + } + + void unpublishAll() { + if (participant_ != nullptr) { + for (const auto& sid : track_sids_) { + if (!sid.empty()) { + participant_->unpublishTrack(sid); + } + } + } + track_sids_.clear(); + } + +private: + LocalParticipant* participant_ = nullptr; + std::vector track_sids_; +}; + +} // namespace + +class RoomEventDeduplicationIntegrationTest : public LiveKitTestBase, public ::testing::WithParamInterface { +protected: + void SetUp() override { + LiveKitTestBase::SetUp(); + if (!config_.available) { + GTEST_SKIP() << "LIVEKIT_URL, LIVEKIT_TOKEN_A, and LIVEKIT_TOKEN_B not set"; + } + } +}; + +TEST_P(RoomEventDeduplicationIntegrationTest, RoomLifecycleDelegateCallbacksFireExactlyOnce) { + const bool single_peer_connection = GetParam(); + + RoomOptions options; + options.auto_subscribe = true; + options.single_peer_connection = single_peer_connection; + + RoomEventCounts observer_counts; + RoomEventCounterDelegate observer_delegate(observer_counts); + + Room observer_room; + observer_room.setDelegate(&observer_delegate); + ASSERT_TRUE(observer_room.connect(config_.url, config_.token_b, options)) << "Observer failed to connect"; + ASSERT_FALSE(observer_room.localParticipant().expired()); + + Room peer_room; + ASSERT_TRUE(peer_room.connect(config_.url, config_.token_a, options)) << "Peer failed to connect"; + ASSERT_FALSE(peer_room.localParticipant().expired()); + + const std::string peer_identity = lockLocalParticipant(peer_room)->identity(); + ASSERT_FALSE(peer_identity.empty()); + + const std::map peer_identity_expected{{peer_identity, 1}}; + ASSERT_TRUE(waitForMapCountAtLeast(observer_counts, peer_identity_expected, 1, kEventWaitTimeout)) + << "Timed out waiting for onParticipantConnected"; + ASSERT_TRUE(waitForParticipant(&observer_room, peer_identity, 10s)) << "Peer not visible to observer room"; + { + const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts); + expectMapCountsExact(snapshot.participant_connected, peer_identity_expected, "onParticipantConnected"); + expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after participant connected"); + } + + const std::string audio_track_name = makeUniqueTrackName("dedupe-audio"); + const std::string video_track_name = makeUniqueTrackName("dedupe-video"); + + auto audio_source = std::make_shared(kDefaultAudioSampleRate, kDefaultAudioChannels, 0); + auto video_source = std::make_shared(kDefaultVideoWidth, kDefaultVideoHeight); + auto audio_track = LocalAudioTrack::createLocalAudioTrack(audio_track_name, audio_source); + auto video_track = LocalVideoTrack::createLocalVideoTrack(video_track_name, video_source); + + TrackPublishOptions audio_opts; + audio_opts.source = TrackSource::SOURCE_MICROPHONE; + TrackPublishOptions video_opts; + video_opts.source = TrackSource::SOURCE_CAMERA; + + auto peer_participant = lockLocalParticipant(peer_room); + PublishedTrackGuard published_tracks(peer_participant.get()); + MediaLoopGuard media_loops; + + ASSERT_NO_THROW(peer_participant->publishTrack(audio_track, audio_opts)); + ASSERT_NE(audio_track->publication(), nullptr); + published_tracks.addTrackSid(audio_track->publication()->sid()); + media_loops.addAudioSource(audio_source); + + ASSERT_NO_THROW(peer_participant->publishTrack(video_track, video_opts)); + ASSERT_NE(video_track->publication(), nullptr); + published_tracks.addTrackSid(video_track->publication()->sid()); + media_loops.addVideoSource(video_source); + + const std::map expected_subscribed_counts{{audio_track_name, 1}, {video_track_name, 1}}; + + ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, expected_subscribed_counts, + &RoomEventCounts::track_subscribed, kEventWaitTimeout)) + << "Timed out waiting for onTrackSubscribed; observed track_subscribed=" + << describeCounts(snapshotCounts(observer_counts).track_subscribed); + + { + const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts); + expectMapCountsExact(snapshot.track_subscribed, expected_subscribed_counts, "onTrackSubscribed"); + expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after track subscribe"); + } + + media_loops.stop(); + published_tracks.unpublishAll(); + + const std::map expected_unsubscribed_counts{{audio_track_name, 1}, {video_track_name, 1}}; + ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, expected_unsubscribed_counts, + &RoomEventCounts::track_unsubscribed, kEventWaitTimeout)) + << "Timed out waiting for onTrackUnsubscribed; observed track_unsubscribed=" + << describeCounts(snapshotCounts(observer_counts).track_unsubscribed); + + { + const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts); + expectMapCountsExact(snapshot.track_unsubscribed, expected_unsubscribed_counts, "onTrackUnsubscribed"); + expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after track unsubscribed"); + } + + peer_room.disconnect(); + + ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, peer_identity_expected, + &RoomEventCounts::participant_disconnected, kEventWaitTimeout)) + << "Timed out waiting for onParticipantDisconnected"; + + { + const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts); + expectMapCountsExact(snapshot.participant_disconnected, peer_identity_expected, "onParticipantDisconnected"); + expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after participant disconnected"); + } + + ASSERT_TRUE(observer_room.disconnect()) << "Observer disconnect failed"; + EXPECT_EQ(snapshotCounts(observer_counts).disconnected, 1) << "onDisconnected should fire exactly once"; + + const RoomEventCountsSnapshot after_disconnect = snapshotCounts(observer_counts); + expectCountsUnchangedAfterGrace(observer_counts, after_disconnect, "after observer disconnect"); + + EXPECT_FALSE(observer_room.disconnect()) << "Second disconnect should be a no-op"; + EXPECT_EQ(snapshotCounts(observer_counts).disconnected, 1) << "onDisconnected must not double-fire"; +} + +INSTANTIATE_TEST_SUITE_P(SingleAndDualPeerConnection, RoomEventDeduplicationIntegrationTest, ::testing::Bool()); + +} // namespace livekit::test diff --git a/src/tests/unit/test_room.cpp b/src/tests/unit/test_room.cpp index 22769930..f514508a 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -24,38 +24,11 @@ #include #include "../common/ffi_utils.h" +#include "../common/room_test_access.h" #include "ffi.pb.h" #include "ffi_client.h" #include "room_proto_converter.h" -namespace livekit { - -struct RoomTestAccess { - 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); - room.onEvent(event); - }); - - const std::scoped_lock guard(room.lock_); - room.connection_state_ = ConnectionState::Connected; - room.room_handle_ = std::make_shared(); - room.listener_id_ = listener_id; - } - - static bool hasRoomHandle(const Room& room) { - const std::scoped_lock guard(room.lock_); - return static_cast(room.room_handle_); - } - - static int listenerId(const Room& room) { - const std::scoped_lock guard(room.lock_); - return room.listener_id_; - } -}; - -} // namespace livekit - namespace livekit::test { class RoomTest : public ::testing::Test { diff --git a/src/tests/unit/test_room_callbacks.cpp b/src/tests/unit/test_room_callbacks.cpp index 71349d5b..ad89be14 100644 --- a/src/tests/unit/test_room_callbacks.cpp +++ b/src/tests/unit/test_room_callbacks.cpp @@ -37,12 +37,20 @@ class RoomCallbackTest : public ::testing::Test { TEST_F(RoomCallbackTest, FrameCallbackRegistrationByTrackNameIsAccepted) { Room room; - EXPECT_NO_THROW(room.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {})); - EXPECT_NO_THROW(room.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {})); + room.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + room.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); EXPECT_NO_THROW(room.clearOnAudioFrameCallback("alice", "mic-main")); EXPECT_NO_THROW(room.clearOnVideoFrameCallback("alice", "cam-main")); } +TEST_F(RoomCallbackTest, TrySetOnAudioReturnsTrueWithoutSubscription) { + // Without a subscribed track, registration succeeds and no reader starts. + Room room; + room.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); + // Re-registering the same key while no reader is active is allowed. + room.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}); +} + TEST_F(RoomCallbackTest, DataCallbackRegistrationReturnsUsableIds) { Room room; diff --git a/src/tests/unit/test_subscription_thread_dispatcher.cpp b/src/tests/unit/test_subscription_thread_dispatcher.cpp index 80b52120..fa035cb9 100644 --- a/src/tests/unit/test_subscription_thread_dispatcher.cpp +++ b/src/tests/unit/test_subscription_thread_dispatcher.cpp @@ -18,14 +18,54 @@ #include #include +#include #include +#include +#include +#include +#include +#include #include #include #include +#include "../common/remote_data_track_test_access.h" + namespace livekit { +namespace { + +using namespace std::chrono_literals; + +/// Minimal Track used to drive audio/video reader startup decisions without a +/// live FFI handle. The SID-skip check runs before any FFI call, so an invalid +/// handle is sufficient to exercise it deterministically. +class FakeMediaTrack : public Track { +public: + FakeMediaTrack(std::string sid, TrackKind kind) + : Track(FfiHandle(0), std::move(sid), "track", kind, StreamState::STATE_ACTIVE, false, true) {} +}; + +/// Minimal frames used to invoke a stored callback directly, so tests can prove +/// which callback a registration slot actually holds. +AudioFrame makeAudioFrame() { return AudioFrame::create(48000, 1, 480); } +VideoFrame makeVideoFrame() { return VideoFrame::create(16, 16, VideoBufferType::RGBA); } + +template +bool waitFor(Predicate predicate, std::chrono::milliseconds timeout) { + const auto start = std::chrono::steady_clock::now(); + while (std::chrono::steady_clock::now() - start < timeout) { + if (predicate()) { + return true; + } + std::this_thread::sleep_for(5ms); + } + return predicate(); +} + +} // namespace + class SubscriptionThreadDispatcherTest : public ::testing::Test { protected: void SetUp() override { livekit::initialize(livekit::LogLevel::Info); } @@ -37,6 +77,8 @@ class SubscriptionThreadDispatcherTest : public ::testing::Test { using DataCallbackKey = SubscriptionThreadDispatcher::DataCallbackKey; using DataCallbackKeyHash = SubscriptionThreadDispatcher::DataCallbackKeyHash; + using ActiveDataReader = SubscriptionThreadDispatcher::ActiveDataReader; + static auto& audioCallbacks(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.audio_callbacks_; } static auto& videoCallbacks(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.video_callbacks_; } static auto& activeReaders(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.active_readers_; } @@ -44,6 +86,32 @@ class SubscriptionThreadDispatcherTest : public ::testing::Test { static auto& activeDataReaders(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.active_data_readers_; } static auto& remoteDataTracks(SubscriptionThreadDispatcher& dispatcher) { return dispatcher.remote_data_tracks_; } static int maxActiveReaders() { return SubscriptionThreadDispatcher::kMaxActiveReaders; } + static bool isSelfThread(std::thread::id id) { return SubscriptionThreadDispatcher::isSelfThread(id); } + static std::size_t activeReaderCount(SubscriptionThreadDispatcher& dispatcher) { + const std::scoped_lock lock(dispatcher.lock_); + return dispatcher.active_readers_.size(); + } + static std::size_t activeDataReaderCount(SubscriptionThreadDispatcher& dispatcher) { + const std::scoped_lock lock(dispatcher.lock_); + return dispatcher.active_data_readers_.size(); + } + + static std::thread extractDataReader(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id) { + const std::scoped_lock lock(dispatcher.lock_); + return dispatcher.extractDataReaderThreadLocked(id); + } + + static void markDataReaderFinishedIfCurrent(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id, + const std::shared_ptr& reader) { + dispatcher.markDataReaderFinishedIfCurrent(id, reader); + } + + static std::thread startDataReader(SubscriptionThreadDispatcher& dispatcher, DataFrameCallbackId id, + const DataCallbackKey& key, const std::shared_ptr& track) { + const std::scoped_lock lock(dispatcher.lock_); + return dispatcher.startDataReaderLocked(id, key, track, + [](const std::vector&, std::optional) {}); + } }; // ============================================================================ @@ -170,23 +238,54 @@ TEST_F(SubscriptionThreadDispatcherTest, ClearNonExistentCallbackIsNoOp) { EXPECT_NO_THROW(dispatcher.clearOnVideoFrameCallback("nobody", "missing")); } -TEST_F(SubscriptionThreadDispatcherTest, OverwriteAudioCallbackKeepsSingleEntry) { +TEST_F(SubscriptionThreadDispatcherTest, OverwriteAudioCallbackStoresTheNewCallback) { SubscriptionThreadDispatcher dispatcher; - std::atomic counter1{0}; - std::atomic counter2{0}; + std::atomic first{0}; + std::atomic second{0}; - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [&counter1](const AudioFrame&) { counter1++; }); - dispatcher.setOnAudioFrameCallback("alice", "mic-main", [&counter2](const AudioFrame&) { counter2++; }); + dispatcher.setOnAudioFrameCallback("alice", "mic-main", [&first](const AudioFrame&) { first++; }); + dispatcher.setOnAudioFrameCallback("alice", "mic-main", [&second](const AudioFrame&) { second++; }); EXPECT_EQ(audioCallbacks(dispatcher).size(), 1u) << "Re-registering with the same key should overwrite, not add"; + + // Invoke what the slot actually holds: size alone would not catch a setter + // that tore down the reader but forgot to install the new callback. + const CallbackKey key{"alice", "mic-main"}; + audioCallbacks(dispatcher)[key].callback(makeAudioFrame()); + EXPECT_EQ(first.load(), 0) << "The replaced callback must not be the one stored"; + EXPECT_EQ(second.load(), 1); } -TEST_F(SubscriptionThreadDispatcherTest, OverwriteVideoCallbackKeepsSingleEntry) { +TEST_F(SubscriptionThreadDispatcherTest, OverwriteVideoCallbackStoresTheNewCallback) { SubscriptionThreadDispatcher dispatcher; - dispatcher.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); - dispatcher.setOnVideoFrameCallback("alice", "cam-main", [](const VideoFrame&, std::int64_t) {}); + std::atomic first{0}; + std::atomic second{0}; + + dispatcher.setOnVideoFrameCallback("alice", "cam-main", [&first](const VideoFrame&, std::int64_t) { first++; }); + dispatcher.setOnVideoFrameCallback("alice", "cam-main", [&second](const VideoFrame&, std::int64_t) { second++; }); EXPECT_EQ(videoCallbacks(dispatcher).size(), 1u); + + const CallbackKey key{"alice", "cam-main"}; + videoCallbacks(dispatcher)[key].legacy_callback(makeVideoFrame(), 0); + EXPECT_EQ(first.load(), 0); + EXPECT_EQ(second.load(), 1); +} + +TEST_F(SubscriptionThreadDispatcherTest, OverwriteAudioCallbackStoresTheNewStreamOptions) { + SubscriptionThreadDispatcher dispatcher; + AudioStream::Options first_opts; + first_opts.capacity = 4; + AudioStream::Options second_opts; + second_opts.capacity = 32; + + dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}, first_opts); + dispatcher.setOnAudioFrameCallback("alice", "mic-main", [](const AudioFrame&) {}, second_opts); + + // The options travel with the callback into the next reader, so a stale copy + // would silently rebuild the stream with the wrong queue behavior. + const CallbackKey key{"alice", "mic-main"}; + EXPECT_EQ(audioCallbacks(dispatcher)[key].options.capacity, 32u); } TEST_F(SubscriptionThreadDispatcherTest, MultipleDistinctCallbacksAreIndependent) { @@ -478,6 +577,408 @@ TEST_F(SubscriptionThreadDispatcherTest, NoRemoteDataTracksInitially) { EXPECT_TRUE(remoteDataTracks(dispatcher).empty()); } +// ============================================================================ +// Data reader replacement: cancellation and finished-state ownership +// ============================================================================ + +TEST_F(SubscriptionThreadDispatcherTest, ActiveDataReaderNotCancelledByDefault) { + auto reader = std::make_shared(); + EXPECT_FALSE(reader->cancelled.load()); + EXPECT_FALSE(reader->finished); +} + +TEST_F(SubscriptionThreadDispatcherTest, ExtractDataReaderMarksCancelledAndRemovesEntry) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + activeDataReaders(dispatcher)[0] = reader; + + auto extracted = extractDataReader(dispatcher, 0); + + EXPECT_TRUE(reader->cancelled.load()) << "Extract must cancel so an in-flight subscribe aborts"; + EXPECT_FALSE(extracted.joinable()) << "No real thread was attached to the seeded reader"; + EXPECT_TRUE(activeDataReaders(dispatcher).empty()); +} + +TEST_F(SubscriptionThreadDispatcherTest, ExtractMissingDataReaderIsNoOp) { + SubscriptionThreadDispatcher dispatcher; + auto extracted = extractDataReader(dispatcher, 42); + EXPECT_FALSE(extracted.joinable()); +} + +TEST_F(SubscriptionThreadDispatcherTest, MarkDataReaderFinishedIfCurrentKeepsMatchingEntry) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + activeDataReaders(dispatcher)[0] = reader; + + markDataReaderFinishedIfCurrent(dispatcher, 0, reader); + + ASSERT_EQ(activeDataReaders(dispatcher).size(), 1u); + EXPECT_EQ(activeDataReaders(dispatcher)[0], reader); + EXPECT_TRUE(reader->finished); +} + +TEST_F(SubscriptionThreadDispatcherTest, MarkDataReaderFinishedIfCurrentLeavesReplacedEntry) { + SubscriptionThreadDispatcher dispatcher; + auto original = std::make_shared(); + auto replacement = std::make_shared(); + activeDataReaders(dispatcher)[0] = replacement; + + // The original reader exited after being replaced; it must not mark the + // newer reader that now owns the same callback id. + markDataReaderFinishedIfCurrent(dispatcher, 0, original); + + ASSERT_EQ(activeDataReaders(dispatcher).size(), 1u); + EXPECT_EQ(activeDataReaders(dispatcher)[0], replacement); + EXPECT_FALSE(replacement->finished); +} + +TEST_F(SubscriptionThreadDispatcherTest, ExtractFinishedDataReaderRemovesEntryAndReturnsJoinableThread) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + reader->finished = true; + reader->thread = std::thread([]() {}); + activeDataReaders(dispatcher)[0] = reader; + + auto extracted = extractDataReader(dispatcher, 0); + + EXPECT_TRUE(reader->cancelled.load()); + EXPECT_TRUE(extracted.joinable()); + EXPECT_TRUE(activeDataReaders(dispatcher).empty()); + extracted.join(); +} + +// ============================================================================ +// SID deduplication: audio/video reader start is skipped for the same SID +// ============================================================================ + +TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameAudioSidDoesNotRestartReader) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + + // Simulate an already-running reader for this subscription. + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + // A duplicate track_subscribed carrying the same SID must be a no-op: no + // extract, no new stream/thread. + auto track = std::make_shared("TR_audio_1", TrackKind::KIND_AUDIO); + dispatcher.handleTrackSubscribed("alice", "mic", track); + + EXPECT_EQ(activeReaderCount(dispatcher), 1u); + EXPECT_EQ(activeReaders(dispatcher)[key].track_sid, "TR_audio_1"); + EXPECT_EQ(activeReaders(dispatcher)[key].audio_stream, nullptr) << "Reader must not have been rebuilt"; +} + +TEST_F(SubscriptionThreadDispatcherTest, DuplicateSubscribeWithSameVideoSidDoesNotRestartReader) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); + + const CallbackKey key{"alice", "cam"}; + activeReaders(dispatcher)[key].track_sid = "TR_video_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + auto track = std::make_shared("TR_video_1", TrackKind::KIND_VIDEO); + dispatcher.handleTrackSubscribed("alice", "cam", track); + + EXPECT_EQ(activeReaderCount(dispatcher), 1u); + EXPECT_EQ(activeReaders(dispatcher)[key].track_sid, "TR_video_1"); + EXPECT_EQ(activeReaders(dispatcher)[key].video_stream, nullptr) << "Reader must not have been rebuilt"; +} + +// ============================================================================ +// setOn* replacement semantics: re-registering for a key with an active reader +// stops that reader in place so the next start binds the new callback. +// ============================================================================ + +TEST_F(SubscriptionThreadDispatcherTest, SetOnAudioWhileReaderActiveReplacesRegistrationAndStopsReader) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + + // Simulate an already-running reader for this subscription. + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + std::atomic replacement_invocations{0}; + dispatcher.setOnAudioFrameCallback("alice", "mic", + [&replacement_invocations](const AudioFrame&) { replacement_invocations++; }); + + EXPECT_EQ(activeReaderCount(dispatcher), 0u) << "The stale reader must be extracted so it stops dispatching to the " + "callback it captured by value"; + ASSERT_EQ(audioCallbacks(dispatcher).size(), 1u); + audioCallbacks(dispatcher)[key].callback(makeAudioFrame()); + EXPECT_EQ(replacement_invocations.load(), 1) << "The replacement callback must be the one now stored"; +} + +TEST_F(SubscriptionThreadDispatcherTest, SetOnVideoWhileReaderActiveReplacesRegistrationAndStopsReader) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); + + const CallbackKey key{"alice", "cam"}; + activeReaders(dispatcher)[key].track_sid = "TR_video_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + std::atomic replacement_invocations{0}; + dispatcher.setOnVideoFrameCallback( + "alice", "cam", [&replacement_invocations](const VideoFrame&, std::int64_t) { replacement_invocations++; }); + + EXPECT_EQ(activeReaderCount(dispatcher), 0u); + ASSERT_EQ(videoCallbacks(dispatcher).size(), 1u); + videoCallbacks(dispatcher)[key].legacy_callback(makeVideoFrame(), 0); + EXPECT_EQ(replacement_invocations.load(), 1) << "The replacement callback must be the one now stored"; +} + +TEST_F(SubscriptionThreadDispatcherTest, SetOnVideoEventWhileReaderActiveReplacesRegistrationAndStopsReader) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); + + const CallbackKey key{"alice", "cam"}; + activeReaders(dispatcher)[key].track_sid = "TR_video_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + // The legacy and event callbacks share one registration slot, so registering + // the event variant must displace the legacy one and stop its reader. + dispatcher.setOnVideoFrameEventCallback("alice", "cam", [](const VideoFrameEvent&) {}); + + EXPECT_EQ(activeReaderCount(dispatcher), 0u); + ASSERT_EQ(videoCallbacks(dispatcher).size(), 1u); + EXPECT_FALSE(static_cast(videoCallbacks(dispatcher)[key].legacy_callback)); + EXPECT_TRUE(static_cast(videoCallbacks(dispatcher)[key].event_callback)); +} + +TEST_F(SubscriptionThreadDispatcherTest, SetOnAudioAfterReplacementRestartsOnNextSubscribe) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + // Replacing extracts the reader, so the SID dedup guard no longer suppresses a + // restart for the same publication -- this is what lets Room rebuild the reader + // against the new callback. + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + ASSERT_EQ(activeReaderCount(dispatcher), 0u); + + // Re-subscribing the same SID now reaches stream construction instead of being + // short-circuited by the guard. The fake track carries an invalid FFI handle, + // so AudioStream::fromTrack throws -- that throw is precisely the evidence + // that startup was attempted rather than skipped. + auto track = std::make_shared("TR_audio_1", TrackKind::KIND_AUDIO); + EXPECT_ANY_THROW(dispatcher.handleTrackSubscribed("alice", "mic", track)); + + EXPECT_EQ(audioCallbacks(dispatcher).size(), 1u); +} + +TEST_F(SubscriptionThreadDispatcherTest, SetOnVideoAfterReplacementRestartsOnNextSubscribe) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); + + const CallbackKey key{"alice", "cam"}; + activeReaders(dispatcher)[key].track_sid = "TR_video_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + dispatcher.setOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); + ASSERT_EQ(activeReaderCount(dispatcher), 0u); + + // As in the audio case, the throw from VideoStream::fromTrack on the invalid + // fake handle is the evidence that the guard no longer short-circuits startup. + auto track = std::make_shared("TR_video_1", TrackKind::KIND_VIDEO); + EXPECT_ANY_THROW(dispatcher.handleTrackSubscribed("alice", "cam", track)); + + EXPECT_EQ(videoCallbacks(dispatcher).size(), 1u); +} + +TEST_F(SubscriptionThreadDispatcherTest, SetOnAudioWithoutReplacementLeavesSidGuardIntact) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + // Counterpart to the test above: with the reader still in place, a duplicate + // subscribe for the same SID is skipped and never reaches stream construction. + auto track = std::make_shared("TR_audio_1", TrackKind::KIND_AUDIO); + EXPECT_NO_THROW(dispatcher.handleTrackSubscribed("alice", "mic", track)); + EXPECT_EQ(activeReaderCount(dispatcher), 1u); +} + +// Distinct from ClearAudioCallbackRemovesRegistration, which clears a key that +// has no reader: this covers clearing while a reader is active. +TEST_F(SubscriptionThreadDispatcherTest, ClearAudioCallbackWithActiveReaderStopsReader) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + dispatcher.clearOnAudioFrameCallback("alice", "mic"); + EXPECT_EQ(activeReaderCount(dispatcher), 0u); + EXPECT_TRUE(audioCallbacks(dispatcher).empty()); +} + +TEST_F(SubscriptionThreadDispatcherTest, ClearVideoCallbackWithActiveReaderStopsReader) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); + + const CallbackKey key{"alice", "cam"}; + activeReaders(dispatcher)[key].track_sid = "TR_video_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + dispatcher.clearOnVideoFrameCallback("alice", "cam"); + EXPECT_EQ(activeReaderCount(dispatcher), 0u); + EXPECT_TRUE(videoCallbacks(dispatcher).empty()); +} + +// The reverse of SetOnVideoEventWhileReaderActiveReplacesRegistrationAndStopsReader: +// the legacy setter must displace a stored event callback, not merge with it. +TEST_F(SubscriptionThreadDispatcherTest, SetOnVideoDisplacesStoredEventCallback) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnVideoFrameEventCallback("alice", "cam", [](const VideoFrameEvent&) {}); + + const CallbackKey key{"alice", "cam"}; + activeReaders(dispatcher)[key].track_sid = "TR_video_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + dispatcher.setOnVideoFrameCallback("alice", "cam", [](const VideoFrame&, std::int64_t) {}); + + EXPECT_EQ(activeReaderCount(dispatcher), 0u); + ASSERT_EQ(videoCallbacks(dispatcher).size(), 1u); + EXPECT_TRUE(static_cast(videoCallbacks(dispatcher)[key].legacy_callback)); + EXPECT_FALSE(static_cast(videoCallbacks(dispatcher)[key].event_callback)); +} + +// Replacement must be scoped to its own key; an unrelated subscription's reader +// is extracted by key, so a bug there would tear down the wrong stream. +TEST_F(SubscriptionThreadDispatcherTest, ReplacementLeavesOtherKeysReadersUntouched) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + dispatcher.setOnAudioFrameCallback("bob", "mic", [](const AudioFrame&) {}); + + const CallbackKey alice{"alice", "mic"}; + const CallbackKey bob{"bob", "mic"}; + activeReaders(dispatcher)[alice].track_sid = "TR_audio_1"; + activeReaders(dispatcher)[bob].track_sid = "TR_audio_2"; + ASSERT_EQ(activeReaderCount(dispatcher), 2u); + + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + + EXPECT_EQ(activeReaderCount(dispatcher), 1u); + EXPECT_EQ(activeReaders(dispatcher).count(alice), 0u); + ASSERT_EQ(activeReaders(dispatcher).count(bob), 1u); + EXPECT_EQ(activeReaders(dispatcher)[bob].track_sid, "TR_audio_2") << "Replacing one key must not disturb another"; + EXPECT_EQ(audioCallbacks(dispatcher).size(), 2u); +} + +// Unsubscribe stops the reader but keeps the registration, so a replacement made +// while unsubscribed is the one that binds on the next subscribe. +TEST_F(SubscriptionThreadDispatcherTest, ReplacementWhileUnsubscribedKeepsRegistrationForNextSubscribe) { + SubscriptionThreadDispatcher dispatcher; + dispatcher.setOnAudioFrameCallback("alice", "mic", [](const AudioFrame&) {}); + + const CallbackKey key{"alice", "mic"}; + activeReaders(dispatcher)[key].track_sid = "TR_audio_1"; + ASSERT_EQ(activeReaderCount(dispatcher), 1u); + + dispatcher.handleTrackUnsubscribed("alice", TrackSource::SOURCE_MICROPHONE, "mic"); + EXPECT_EQ(activeReaderCount(dispatcher), 0u); + EXPECT_EQ(audioCallbacks(dispatcher).size(), 1u) << "Unsubscribe must preserve the registration"; + + std::atomic replacement_invocations{0}; + dispatcher.setOnAudioFrameCallback("alice", "mic", + [&replacement_invocations](const AudioFrame&) { replacement_invocations++; }); + ASSERT_EQ(audioCallbacks(dispatcher).size(), 1u); + audioCallbacks(dispatcher)[key].callback(makeAudioFrame()); + EXPECT_EQ(replacement_invocations.load(), 1); +} + +// ============================================================================ +// Self-join detection +// ============================================================================ + +TEST_F(SubscriptionThreadDispatcherTest, IsSelfThreadIdentifiesTheCallingThread) { + EXPECT_TRUE(isSelfThread(std::this_thread::get_id())); + EXPECT_FALSE(isSelfThread(std::thread::id{})) << "A default-constructed id must never match a running thread"; + + std::thread other([]() {}); + const auto other_id = other.get_id(); + other.join(); + EXPECT_FALSE(isSelfThread(other_id)); +} + +// ============================================================================ +// SID deduplication: data reader start is skipped for the same SID and +// replaced (stopping the previous reader) for a new SID +// ============================================================================ + +TEST_F(SubscriptionThreadDispatcherTest, DuplicateDataPublishWithSameSidDoesNotRestartReader) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + reader->remote_track = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); + activeDataReaders(dispatcher)[7] = reader; + + auto incoming = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); + auto old_thread = startDataReader(dispatcher, 7, DataCallbackKey{"alice", "foo"}, incoming); + + EXPECT_FALSE(old_thread.joinable()); + EXPECT_EQ(activeDataReaderCount(dispatcher), 1u); + EXPECT_EQ(activeDataReaders(dispatcher)[7], reader) << "Same-SID publish must not replace the reader"; + EXPECT_FALSE(reader->cancelled.load()) << "A skipped reader must not be cancelled"; +} + +TEST_F(SubscriptionThreadDispatcherTest, FinishedDataReaderWithSameSidIsReplaced) { + SubscriptionThreadDispatcher dispatcher; + auto reader = std::make_shared(); + reader->remote_track = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); + reader->finished = true; + activeDataReaders(dispatcher)[7] = reader; + + auto incoming = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); + auto old_thread = startDataReader(dispatcher, 7, DataCallbackKey{"alice", "foo"}, incoming); + if (old_thread.joinable()) { + old_thread.join(); + } + + EXPECT_TRUE(reader->cancelled.load()) << "Finished reader must be extracted before replacement"; + EXPECT_TRUE(waitFor( + [&] { + return activeDataReaderCount(dispatcher) == 1u && activeDataReaders(dispatcher)[7] != reader && + activeDataReaders(dispatcher)[7]->finished; + }, + 2s)); + + dispatcher.stopAll(); + EXPECT_TRUE(activeDataReaders(dispatcher).empty()); +} + +TEST_F(SubscriptionThreadDispatcherTest, RepublishWithNewDataSidStopsPreviousReader) { + SubscriptionThreadDispatcher dispatcher; + auto previous = std::make_shared(); + previous->remote_track = RemoteDataTrackTestAccess::create({"foo", "TR_data_1", false}, "alice"); + activeDataReaders(dispatcher)[7] = previous; + + // A republish under the same (participant, name) but a NEW SID must stop the + // previous reader and start a fresh one. + auto republished = RemoteDataTrackTestAccess::create({"foo", "TR_data_2", false}, "alice"); + auto old_thread = startDataReader(dispatcher, 7, DataCallbackKey{"alice", "foo"}, republished); + if (old_thread.joinable()) { + old_thread.join(); + } + + EXPECT_TRUE(previous->cancelled.load()) << "Previous reader must be cancelled on republish"; + + // The replacement reader has an invalid FFI handle, so its subscribe fails + // fast and marks itself finished while the dispatcher keeps ownership. + EXPECT_TRUE(waitFor( + [&] { return activeDataReaderCount(dispatcher) == 1u && activeDataReaders(dispatcher)[7]->finished; }, 2s)); + + dispatcher.stopAll(); + EXPECT_TRUE(activeDataReaders(dispatcher).empty()); +} + // ============================================================================ // Data track destruction safety // ============================================================================