fix bad test

stephen-derosa · stephen-derosa · commit af9bed761e89 · 2026-05-12T13:51:54.000-06:00
diff --git a/include/livekit/room.h b/include/livekit/room.h
@@ -306,7 +306,6 @@ class LIVEKIT_API Room {
 
   mutable std::mutex lock_;
   ConnectionState connection_state_ = ConnectionState::Disconnected;
-  bool connect_in_progress_ = false;
   RoomDelegate* delegate_ = nullptr; // Not owned
   RoomInfoData room_info_;
   std::shared_ptr<FfiHandle> room_handle_;
diff --git a/src/room.cpp b/src/room.cpp
@@ -16,9 +16,7 @@
 
 #include "livekit/room.h"
 
-#include <chrono>
 #include <functional>
-#include <thread>
 
 #include "data_track.pb.h"
 #include "ffi.pb.h"
@@ -111,7 +109,6 @@ bool Room::Connect(const std::string& url, const std::string& token, const RoomO
     if (connection_state_ != ConnectionState::Disconnected) {
       throw std::runtime_error("already connected");
     }
-    connect_in_progress_ = true;
     connection_state_ = ConnectionState::Reconnecting;
   }
 
@@ -190,16 +187,11 @@ bool Room::Connect(const std::string& url, const std::string& token, const RoomO
       connection_state_ = ConnectionState::Connected;
     }
 
-    {
-      const std::scoped_lock<std::mutex> g(lock_);
-      connect_in_progress_ = false;
-    }
     return true;
   } catch (const std::exception& e) {
     int listener_to_remove = 0;
     {
       const std::scoped_lock<std::mutex> g(lock_);
-      connect_in_progress_ = false;
       connection_state_ = ConnectionState::Disconnected;
       if (listener_id_ == listenerId) {
         listener_to_remove = listener_id_;
@@ -361,39 +353,28 @@ void Room::OnEvent(const FfiEvent& event) {
   // Take a snapshot of the delegate under lock, but do NOT call it under the
   // lock.
   RoomDelegate* delegate_snapshot = nullptr;
-  bool connect_in_progress = false;
   {
     const std::scoped_lock<std::mutex> guard(lock_);
     delegate_snapshot = delegate_;
-    connect_in_progress = connect_in_progress_;
   }
 
   // First, handle RPC method invocations (not part of RoomEvent).
   if (event.message_case() == FfiEvent::kRpcMethodInvocation) {
     const auto& rpc = event.rpc_method_invocation();
 
     LocalParticipant* lp = nullptr;
-    bool local_participant_missing_during_connect = false;
     {
       const std::scoped_lock<std::mutex> guard(lock_);
       if (!local_participant_) {
-        local_participant_missing_during_connect = connect_in_progress_;
-      } else {
-        auto local_handle = local_participant_->ffiHandleId();
-        if (local_handle == INVALID_HANDLE ||
-            rpc.local_participant_handle() != static_cast<std::uint64_t>(local_handle)) {
-          // RPC is not targeted at this room's local participant; ignore.
-          return;
-        }
-        lp = local_participant_.get();
+        return;
       }
-    }
-    if (local_participant_missing_during_connect) {
-      LK_LOG_WARN("dropping RPC invocation while Room::Connect is still publishing local participant state");
-      return;
-    }
-    if (!lp) {
-      return;
+      auto local_handle = local_participant_->ffiHandleId();
+      if (local_handle == INVALID_HANDLE ||
+          rpc.local_participant_handle() != static_cast<std::uint64_t>(local_handle)) {
+        // RPC is not targeted at this room's local participant; ignore.
+        return;
+      }
+      lp = local_participant_.get();
     }
 
     // Call outside the lock to avoid deadlocks / re-entrancy issues.
@@ -408,43 +389,13 @@ void Room::OnEvent(const FfiEvent& event) {
       const proto::RoomEvent& re = event.room_event();
 
       // Check if this event is for our room handle
-      bool missing_room_handle = false;
-      bool missing_local_participant = false;
-      ConnectionState connection_state = ConnectionState::Disconnected;
       {
         const std::scoped_lock<std::mutex> guard(lock_);
-        connection_state = connection_state_;
-        if (!room_handle_) {
-          missing_room_handle = true;
-        } else if (re.room_handle() != static_cast<std::uint64_t>(room_handle_->get())) {
+        if (!room_handle_ || re.room_handle() != static_cast<std::uint64_t>(room_handle_->get())) {
           return;
-        } else if (!local_participant_) {
-          missing_local_participant = true;
         }
       }
 
-      if (missing_room_handle) {
-        LK_LOG_WARN(
-            "dropping room event {} before room handle is initialized; "
-            "connect_in_progress={}, connection_state={}",
-            static_cast<int>(re.message_case()), connect_in_progress, static_cast<int>(connection_state));
-        return;
-      }
-
-      if (missing_local_participant) {
-        LK_LOG_ERROR(
-            "room event {} matched the room handle before local participant state was initialized; "
-            "room state is incomplete",
-            static_cast<int>(re.message_case()));
-      }
-
-      if (connect_in_progress && delegate_snapshot) {
-        LK_LOG_WARN(
-            "room event {} arrived while Room::Connect is still in progress; "
-            "any delegate callback from this event will run before Connect() returns",
-            static_cast<int>(re.message_case()));
-      }
-
       switch (re.message_case()) {
         case proto::RoomEvent::kParticipantConnected: {
           std::shared_ptr<RemoteParticipant> new_participant;
@@ -1189,7 +1140,6 @@ void Room::OnEvent(const FfiEvent& event) {
             listener_id_ = 0;
 
             // Reset connection state
-            connect_in_progress_ = false;
             connection_state_ = ConnectionState::Disconnected;
 
             // Move state out for cleanup outside lock
diff --git a/src/tests/integration/test_late_join_track_publication.cpp b/src/tests/integration/test_late_join_track_publication.cpp
@@ -271,14 +271,9 @@ class PublishedTrackGuard {
   std::vector<std::shared_ptr<LocalDataTrack>> data_tracks_;
 };
 
-bool hasExpectedMediaCallbacks(const LateJoinPublicationState& state,
-                               const std::vector<ExpectedPublication>& expected_media) {
+bool hasExpectedMediaSubscriptions(const LateJoinPublicationState& state,
+                                   const std::vector<ExpectedPublication>& expected_media) {
   for (const auto& expected : expected_media) {
-    const auto published_it = state.published_media_tracks.find(expected.name);
-    if (published_it == state.published_media_tracks.end() || published_it->second != expected.kind) {
-      return false;
-    }
-
     const auto subscribed_it = state.subscribed_media_tracks.find(expected.name);
     if (subscribed_it == state.subscribed_media_tracks.end() || subscribed_it->second != expected.kind) {
       return false;
@@ -401,38 +396,28 @@ TEST_P(LateJoinTrackPublicationIntegrationTest, ConsumerReceivesAlreadyPublished
 
   {
     std::unique_lock<std::mutex> lock(state.mutex);
+    // Pre-existing media publications are delivered in the Connect snapshot, not as
+    // TrackPublished room events. The late-joiner should still receive TrackSubscribed
+    // callbacks once auto-subscribe attaches to those snapshot publications.
     const bool got_expected =
-        state.cv.wait_for(lock, kWaitTimeout, [&]() { return hasExpectedMediaCallbacks(state, expected_media); });
-    EXPECT_TRUE(got_expected) << "Timed out waiting for late-join audio publication/subscription events\n"
+        state.cv.wait_for(lock, kWaitTimeout, [&]() { return hasExpectedMediaSubscriptions(state, expected_media); });
+    EXPECT_TRUE(got_expected) << "Timed out waiting for late-join audio subscription events\n"
                               << "Published media events: " << describeMediaTracks(state.published_media_tracks) << "\n"
                               << "Subscribed media events: " << describeMediaTracks(state.subscribed_media_tracks);
   }
 
-  std::map<std::string, TrackKind> published_media_snapshot;
   std::map<std::string, TrackKind> subscribed_media_snapshot;
-  std::map<std::string, int> published_media_counts;
   std::map<std::string, int> subscribed_media_counts;
   std::vector<std::string> invariant_failures;
   {
     std::lock_guard<std::mutex> lock(state.mutex);
-    published_media_snapshot = state.published_media_tracks;
     subscribed_media_snapshot = state.subscribed_media_tracks;
-    published_media_counts = state.published_media_counts;
     subscribed_media_counts = state.subscribed_media_counts;
     invariant_failures = state.invariant_failures;
   }
   EXPECT_TRUE(invariant_failures.empty()) << describeInvariantFailures(invariant_failures);
 
   for (const auto& expected : expected_media) {
-    const auto published_it = published_media_snapshot.find(expected.name);
-    EXPECT_NE(published_it, published_media_snapshot.end())
-        << "Missing onTrackPublished event for " << expected.name
-        << "; received: " << describeMediaTracks(published_media_snapshot);
-    if (published_it != published_media_snapshot.end()) {
-      EXPECT_EQ(published_it->second, expected.kind) << "Published track kind mismatch for " << expected.name;
-    }
-    EXPECT_EQ(published_media_counts[expected.name], 1) << "Unexpected onTrackPublished count for " << expected.name;
-
     const auto subscribed_it = subscribed_media_snapshot.find(expected.name);
     EXPECT_NE(subscribed_it, subscribed_media_snapshot.end())
         << "Missing onTrackSubscribed event for " << expected.name
@@ -516,38 +501,28 @@ TEST_P(LateJoinTrackPublicationIntegrationTest, ConsumerReceivesAlreadyPublished
 
   {
     std::unique_lock<std::mutex> lock(state.mutex);
+    // Pre-existing media publications are delivered in the Connect snapshot, not as
+    // TrackPublished room events. The late-joiner should still receive TrackSubscribed
+    // callbacks once auto-subscribe attaches to those snapshot publications.
     const bool got_expected =
-        state.cv.wait_for(lock, kWaitTimeout, [&]() { return hasExpectedMediaCallbacks(state, expected_media); });
-    EXPECT_TRUE(got_expected) << "Timed out waiting for late-join video publication/subscription events\n"
+        state.cv.wait_for(lock, kWaitTimeout, [&]() { return hasExpectedMediaSubscriptions(state, expected_media); });
+    EXPECT_TRUE(got_expected) << "Timed out waiting for late-join video subscription events\n"
                               << "Published media events: " << describeMediaTracks(state.published_media_tracks) << "\n"
                               << "Subscribed media events: " << describeMediaTracks(state.subscribed_media_tracks);
   }
 
-  std::map<std::string, TrackKind> published_media_snapshot;
   std::map<std::string, TrackKind> subscribed_media_snapshot;
-  std::map<std::string, int> published_media_counts;
   std::map<std::string, int> subscribed_media_counts;
   std::vector<std::string> invariant_failures;
   {
     std::lock_guard<std::mutex> lock(state.mutex);
-    published_media_snapshot = state.published_media_tracks;
     subscribed_media_snapshot = state.subscribed_media_tracks;
-    published_media_counts = state.published_media_counts;
     subscribed_media_counts = state.subscribed_media_counts;
     invariant_failures = state.invariant_failures;
   }
   EXPECT_TRUE(invariant_failures.empty()) << describeInvariantFailures(invariant_failures);
 
   for (const auto& expected : expected_media) {
-    const auto published_it = published_media_snapshot.find(expected.name);
-    EXPECT_NE(published_it, published_media_snapshot.end())
-        << "Missing onTrackPublished event for " << expected.name
-        << "; received: " << describeMediaTracks(published_media_snapshot);
-    if (published_it != published_media_snapshot.end()) {
-      EXPECT_EQ(published_it->second, expected.kind) << "Published track kind mismatch for " << expected.name;
-    }
-    EXPECT_EQ(published_media_counts[expected.name], 1) << "Unexpected onTrackPublished count for " << expected.name;
-
     const auto subscribed_it = subscribed_media_snapshot.find(expected.name);
     EXPECT_NE(subscribed_it, subscribed_media_snapshot.end())
         << "Missing onTrackSubscribed event for " << expected.name
