more E2E tests

Author: stephen-derosa

include/livekit/remote_data_track.h

@@ -95,6 +95,16 @@ class RemoteDataTrack {
 #ifdef LIVEKIT_TEST_ACCESS
   /// Test-only accessor for exercising lower-level FFI subscription paths.
   uintptr_t testFfiHandleId() const noexcept { return ffiHandleId(); }
+
+  /// @brief Test-only factory that builds a RemoteDataTrack with no live FFI
+  /// handle, used to exercise dispatcher routing without a real subscription.
+  /// @param info Metadata (name, sid, e2ee) to expose via info().
+  /// @param publisher_identity Identity to expose via publisherIdentity().
+  /// @return A RemoteDataTrack whose handle is invalid, so subscribe() fails
+  ///         fast without touching the FFI.
+  static std::shared_ptr<RemoteDataTrack> createForTest(DataTrackInfo info, std::string publisher_identity) {
+    return std::shared_ptr<RemoteDataTrack>(new RemoteDataTrack(std::move(info), std::move(publisher_identity)));
+  }
 #endif
 
   /// Subscribe to this remote data track.
@@ -109,6 +119,12 @@ class RemoteDataTrack {
 
   explicit RemoteDataTrack(const proto::OwnedRemoteDataTrack& owned);
 
+#ifdef LIVEKIT_TEST_ACCESS
+  /// Test-only constructor that bypasses the FFI, leaving an invalid handle.
+  RemoteDataTrack(DataTrackInfo info, std::string publisher_identity)
+      : handle_(0), info_(std::move(info)), publisher_identity_(std::move(publisher_identity)) {}
+#endif
+
   uintptr_t ffiHandleId() const noexcept { return handle_.get(); }
   /// RAII wrapper for the Rust-owned FFI resource.
   FfiHandle handle_;

include/livekit/subscription_thread_dispatcher.h

@@ -17,6 +17,7 @@
 #pragma once
 
 #include <atomic>
+#include <cstddef>
 #include <cstdint>
 #include <functional>
 #include <memory>
@@ -233,6 +234,14 @@ class LIVEKIT_API SubscriptionThreadDispatcher {
   /// thread survives beyond the lifetime of the owning @ref Room.
   void stopAll();
 
+  /// @brief Number of active audio/video reader threads. Intended for tests.
+  /// @return Count of currently active audio and video reader slots.
+  LIVEKIT_INTERNAL_API std::size_t activeReaderCount() const;
+
+  /// @brief Number of active data track reader threads. Intended for tests.
+  /// @return Count of currently active data reader slots.
+  LIVEKIT_INTERNAL_API std::size_t activeDataReaderCount() const;
+
 private:
   friend class SubscriptionThreadDispatcherTest;
 

src/subscription_thread_dispatcher.cpp

@@ -339,6 +339,16 @@ void SubscriptionThreadDispatcher::stopAll() {
   LK_LOG_DEBUG("Stopped {} subscription reader threads", threads.size());
 }
 
+std::size_t SubscriptionThreadDispatcher::activeReaderCount() const {
+  const std::scoped_lock<std::mutex> lock(lock_);
+  return active_readers_.size();
+}
+
+std::size_t SubscriptionThreadDispatcher::activeDataReaderCount() const {
+  const std::scoped_lock<std::mutex> lock(lock_);
+  return active_data_readers_.size();
+}
+
 std::thread SubscriptionThreadDispatcher::extractReaderThreadLocked(const CallbackKey& key) {
   auto it = active_readers_.find(key);
   if (it == active_readers_.end()) {

src/tests/integration/test_data_track.cpp

@@ -468,6 +468,101 @@ 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<TestRoomConnectionOptions> 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<int> frames_received{0};
+  std::mutex payload_mutex;
+  std::vector<std::uint8_t> last_payload;
+  subscriber_room->addOnDataFrameCallback(publisher_identity, track_name,
+                                          [&](const std::vector<std::uint8_t>& payload, std::optional<std::uint64_t>) {
+                                            {
+                                              const std::scoped_lock<std::mutex> lock(payload_mutex);
+                                              last_payload = payload;
+                                            }
+                                            frames_received.fetch_add(1);
+                                          });
+
+  auto publish_with_retry = [&](const std::string& name) -> std::shared_ptr<LocalDataTrack> {
+    std::shared_ptr<LocalDataTrack> 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";
+  const std::string second_sid = remotes.back()->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<std::mutex> lock(payload_mutex);
+    EXPECT_EQ(last_payload, second_frame.payload) << "Callback delivered stale payload after republish";
+  }
+
+  second_track->unpublishDataTrack();
+}
+
 TEST_F(DataTrackE2ETest, SubscribeAfterUnpublishReportsTerminalError) {
   const auto track_name = makeTrackName("subscribe_after_unpublish");