SubscriptionThreadDispatcher: guard against duplicate room events, duplicate room events test

Author: stephen-derosa

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,11 @@ class RemoteDataTrack {
 
 private:
   friend class Room;
+#ifdef LIVEKIT_TEST_ACCESS
+  friend struct RemoteDataTrackTestAccess;
+#endif
 
-  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.

include/livekit/subscription_thread_dispatcher.h

@@ -16,6 +16,8 @@
 
 #pragma once
 
+#include <atomic>
+#include <cstddef>
 #include <cstdint>
 #include <functional>
 #include <memory>
@@ -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,
@@ -151,27 +155,33 @@ class LIVEKIT_API SubscriptionThreadDispatcher {
   /// @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>& 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.
@@ -259,6 +269,9 @@ class LIVEKIT_API SubscriptionThreadDispatcher {
     std::shared_ptr<AudioStream> audio_stream;
     std::shared_ptr<VideoStream> 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;
   };
 
   /// Compound lookup key for a remote participant identity and data track name.
@@ -289,6 +302,9 @@ class LIVEKIT_API SubscriptionThreadDispatcher {
   /// Active read-side resources for one data track stream subscription.
   struct ActiveDataReader {
     std::shared_ptr<RemoteDataTrack> 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<bool> cancelled{false};
     std::mutex sub_mutex;
     std::shared_ptr<DataTrackStream> stream; // guarded by sub_mutex
     std::thread thread;
@@ -333,18 +349,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<RemoteDataTrack>& track, const DataFrameCallback& cb);
 
+  /// Remove @p reader from @ref active_data_readers_ if the slot for @p id
+  /// still refers to it.  Called by the reader thread itself when it exits
+  /// after a failed or cancelled subscription so it does not leave a stale
+  /// entry behind.  Acquires @ref lock_.
+  void eraseDataReaderIfCurrent(DataFrameCallbackId id, const std::shared_ptr<ActiveDataReader>& reader);
+
   /// Protects callback registration maps and active reader state.
   mutable std::mutex lock_;
 

src/subscription_thread_dispatcher.cpp

@@ -259,6 +259,8 @@ 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<std::mutex> sub_guard(reader->sub_mutex);
           if (reader->stream) {
@@ -312,6 +314,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<std::mutex> sub_guard(reader->sub_mutex);
         if (reader->stream) {
@@ -397,6 +401,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<int>(active_readers_.size()) >= kMaxActiveReaders) {
@@ -415,6 +429,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)
@@ -454,6 +469,16 @@ std::thread SubscriptionThreadDispatcher::startVideoReaderLocked(const CallbackK
                                                                  const std::shared_ptr<Track>& 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<int>(active_readers_.size()) >= kMaxActiveReaders) {
@@ -472,6 +497,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;
@@ -522,6 +548,8 @@ std::thread SubscriptionThreadDispatcher::extractDataReaderThreadLocked(DataFram
   }
   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<std::mutex> guard(reader->sub_mutex);
     if (reader->stream) {
@@ -531,28 +559,36 @@ 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<std::mutex> guard(reader->sub_mutex);
-        if (reader->stream) {
-          reader->stream->close();
-        }
-      }
-      return std::move(reader->thread);
-    }
+void SubscriptionThreadDispatcher::eraseDataReaderIfCurrent(DataFrameCallbackId id,
+                                                            const std::shared_ptr<ActiveDataReader>& reader) {
+  const std::scoped_lock<std::mutex> 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;
   }
-  return {};
+  // Detach so the reader can be destroyed by its own thread's final shared_ptr
+  // release without tripping std::thread's joinable-at-destruction check. The
+  // thread is already returning when this runs, so nothing is left to join.
+  if (reader->thread.joinable()) {
+    reader->thread.detach();
+  }
+  active_data_readers_.erase(it);
 }
 
 std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbackId id, const DataCallbackKey& key,
                                                                 const std::shared_ptr<RemoteDataTrack>& track,
                                                                 const DataFrameCallback& cb) {
+  auto existing = active_data_readers_.find(id);
+  if (existing != active_data_readers_.end() && 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<int>(active_readers_.size()) + static_cast<int>(active_data_readers_.size());
@@ -571,7 +607,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<DataTrackStream> stream;
     auto subscribe_result = track->subscribe();
@@ -581,13 +617,21 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac
           "Failed to subscribe to data track \"{}\" from \"{}\": code={} "
           "message={}",
           track_name, identity, static_cast<std::uint32_t>(error.code), error.message);
+      eraseDataReaderIfCurrent(id, reader);
       return;
     }
     stream = subscribe_result.value();
     LK_LOG_INFO("Data reader thread: subscribed to \"{}\" track=\"{}\"", identity, track_name);
 
     {
       const std::scoped_lock<std::mutex> guard(reader->sub_mutex);
+      // A replacement or teardown may have cancelled this reader while the
+      // subscribe was in flight. Close the fresh stream and bail so we do not
+      // leave a second live subscription behind.
+      if (reader->cancelled.load()) {
+        stream->close();
+        return;
+      }
       reader->stream = stream;
     }
 
@@ -606,6 +650,9 @@ std::thread SubscriptionThreadDispatcher::startDataReaderLocked(DataFrameCallbac
           "\"{}\": code={} message={}",
           track_name, identity, static_cast<std::uint32_t>(error->code), error->message);
     }
+    // Clean our own slot if the stream ended on its own (server EOS) and no
+    // extract/teardown already claimed it. A no-op when we were extracted.
+    eraseDataReaderIfCurrent(id, reader);
     LK_LOG_INFO("Data reader thread exiting for \"{}\" track=\"{}\"", identity, track_name);
   });
   // NOLINTEND(bugprone-lambda-function-name)