Implementation of teardown feature for bugfix

Author: alan-george-lk

include/livekit/room.h

@@ -331,6 +331,7 @@ class LIVEKIT_API Room {
 
 private:
   friend class RoomCallbackTest;
+  friend struct RoomTestAccess;
 
   mutable std::mutex lock_;
   ConnectionState connection_state_ = ConnectionState::Disconnected;
@@ -355,5 +356,10 @@ class LIVEKIT_API Room {
   int listener_id_{0};
 
   void onEvent(const proto::FfiEvent& event);
+
+  // Tracks local teardown independently from connection_state_ to handle all teardown cases.
+  bool teardown_started_{false};
+  // Shared teardown path for explicit disconnect, server disconnect, EOS, and destruction.
+  bool teardown(bool disconnect_ffi, DisconnectReason reason, bool notify_delegate);
 };
 } // namespace livekit

src/room.cpp

@@ -102,6 +102,7 @@ bool Room::connect(const std::string& url, const std::string& token, const RoomO
     if (connection_state_ != ConnectionState::Disconnected) {
       throw std::runtime_error("already connected");
     }
+    teardown_started_ = false;
     connection_state_ = ConnectionState::Reconnecting;
   }
 
@@ -212,7 +213,10 @@ bool Room::connect(const std::string& url, const std::string& token, const RoomO
 
 bool Room::disconnect(DisconnectReason reason) {
   TRACE_EVENT0("livekit", "Room::disconnect");
+  return teardown(true, reason, true);
+}
 
+bool Room::teardown(bool disconnect_ffi, DisconnectReason reason, bool notify_delegate) {
   std::shared_ptr<FfiHandle> handle;
   RoomDelegate* delegate_snapshot = nullptr;
   std::shared_ptr<LocalParticipant> local_participant_to_cleanup;
@@ -224,58 +228,81 @@ bool Room::disconnect(DisconnectReason reason) {
 
   {
     const std::scoped_lock<std::mutex> g(lock_);
-    if (connection_state_ == ConnectionState::Disconnected) {
-      // Already torn down (or never connected). Nothing to do.
+    const bool has_room_state = connection_state_ != ConnectionState::Disconnected || listener_id_ != 0 ||
+                                room_handle_ || local_participant_ || !remote_participants_.empty();
+    if (teardown_started_ || !has_room_state) {
       return false;
     }
-    handle = room_handle_;
+    teardown_started_ = true;
+    handle = std::move(room_handle_);
     delegate_snapshot = delegate_;
-    // Take ownership of everything under the lock so the kEos handler (which
-    // also tries to move it out) loses any race here — only one teardown
-    // path operates on this state.
     local_participant_to_cleanup = std::move(local_participant_);
     remote_participants_to_clear = std::move(remote_participants_);
     e2ee_manager_to_clear = std::move(e2ee_manager_);
     text_stream_readers_to_clear = std::move(text_stream_readers_);
     byte_stream_readers_to_clear = std::move(byte_stream_readers_);
     listener_to_remove = listener_id_;
     listener_id_ = 0;
-    room_handle_.reset();
-    // Flip state immediately so the in-flight Disconnected room-event we'll
-    // get back doesn't double-fire onDisconnected. Mirrors Python's
-    // Room.disconnect()
     connection_state_ = ConnectionState::Disconnected;
   }
 
-  // Drain in-flight RPC handlers BEFORE telling Rust to tear down the room.
-  // Mirrors client-sdk-python's Room.disconnect() ordering
+  bool teardown_ok = true;
   if (local_participant_to_cleanup) {
-    local_participant_to_cleanup->shutdown();
+    try {
+      local_participant_to_cleanup->shutdown();
+    } catch (const std::exception& e) {
+      LK_LOG_ERROR("Room teardown: local participant shutdown failed: {}", e.what());
+      teardown_ok = false;
+    } catch (...) {
+      LK_LOG_ERROR("Room teardown: local participant shutdown failed: unknown exception");
+      teardown_ok = false;
+    }
   }
 
-  // Tell the FFI to close the room and wait for the callback. If this fails
-  // we still complete local-side teardown below
-  bool ffi_ok = true;
-  if (handle) {
+  if (disconnect_ffi && handle && handle->valid()) {
     try {
       FfiClient::instance().disconnectAsync(handle->get(), reason).get();
     } catch (const std::exception& e) {
-      LK_LOG_ERROR("Room::disconnect: FFI disconnect failed (continuing local teardown): {}", e.what());
-      ffi_ok = false;
+      LK_LOG_ERROR("Room teardown: FFI disconnect failed (continuing local teardown): {}", e.what());
+      teardown_ok = false;
+    } catch (...) {
+      LK_LOG_ERROR("Room teardown: FFI disconnect failed (continuing local teardown): unknown exception");
+      teardown_ok = false;
     }
   }
 
-  // Stop dispatcher so no track callbacks fire mid-teardown.
   if (subscription_thread_dispatcher_) {
-    subscription_thread_dispatcher_->stopAll();
+    try {
+      subscription_thread_dispatcher_->stopAll();
+    } catch (const std::exception& e) {
+      LK_LOG_ERROR("Room teardown: subscription shutdown failed: {}", e.what());
+      teardown_ok = false;
+    } catch (...) {
+      LK_LOG_ERROR("Room teardown: subscription shutdown failed: unknown exception");
+      teardown_ok = false;
+    }
   }
 
   if (listener_to_remove != 0) {
-    FfiClient::instance().removeListener(listener_to_remove);
+    try {
+      FfiClient::instance().removeListener(listener_to_remove);
+    } catch (const std::exception& e) {
+      LK_LOG_ERROR("Room teardown: listener removal failed: {}", e.what());
+      teardown_ok = false;
+    } catch (...) {
+      LK_LOG_ERROR("Room teardown: listener removal failed: unknown exception");
+      teardown_ok = false;
+    }
   }
 
-  // Fire onDisconnected exactly once, with the reason the caller passed.
-  if (delegate_snapshot) {
+  local_participant_to_cleanup.reset();
+  remote_participants_to_clear.clear();
+  e2ee_manager_to_clear.reset();
+  text_stream_readers_to_clear.clear();
+  byte_stream_readers_to_clear.clear();
+  handle.reset();
+
+  if (notify_delegate && delegate_snapshot) {
     DisconnectedEvent ev;
     ev.reason = reason;
     try {
@@ -287,9 +314,7 @@ bool Room::disconnect(DisconnectReason reason) {
     }
   }
 
-  // Moved-out state (local participant, remote participants, e2ee manager,
-  // stream readers) destructs here, releasing FFI handles.
-  return ffi_ok;
+  return teardown_ok;
 }
 
 RoomInfoData Room::roomInfo() const {
@@ -1168,22 +1193,8 @@ void Room::onEvent(const FfiEvent& event) {
           break;
         }
         case proto::RoomEvent::kDisconnected: {
-          // If disconnect() was driven from our side, it already flipped state
-          // to Disconnected and fired the delegate; skip the duplicate here.
-          bool already_disconnected = false;
-          {
-            const std::scoped_lock<std::mutex> guard(lock_);
-            already_disconnected = (connection_state_ == ConnectionState::Disconnected);
-            connection_state_ = ConnectionState::Disconnected;
-          }
-          if (already_disconnected) {
-            break;
-          }
-          DisconnectedEvent ev;
-          ev.reason = toDisconnectReason(re.disconnected().reason());
-          if (delegate_snapshot) {
-            delegate_snapshot->onDisconnected(*this, ev);
-          }
+          const auto reason = toDisconnectReason(re.disconnected().reason());
+          (void)teardown(false, reason, true);
           break;
         }
         case proto::RoomEvent::kReconnecting: {
@@ -1208,56 +1219,7 @@ void Room::onEvent(const FfiEvent& event) {
           break;
         }
         case proto::RoomEvent::kEos: {
-          if (subscription_thread_dispatcher_) {
-            subscription_thread_dispatcher_->stopAll();
-          }
-
-          int listener_to_remove = 0;
-
-          // Move state out of lock scope before destroying to avoid holding lock
-          // during potentially long destructors
-          std::shared_ptr<LocalParticipant> old_local_participant;
-          std::unordered_map<std::string, std::shared_ptr<RemoteParticipant>> old_remote_participants;
-          std::shared_ptr<FfiHandle> old_room_handle;
-          std::shared_ptr<E2EEManager> old_e2ee_manager;
-          std::unordered_map<std::string, std::shared_ptr<TextStreamReader>> old_text_readers;
-          std::unordered_map<std::string, std::shared_ptr<ByteStreamReader>> old_byte_readers;
-
-          {
-            const std::scoped_lock<std::mutex> guard(lock_);
-            listener_to_remove = listener_id_;
-            listener_id_ = 0;
-
-            // Reset connection state
-            connection_state_ = ConnectionState::Disconnected;
-
-            // Move state out for cleanup outside lock
-            old_local_participant = std::move(local_participant_);
-            old_remote_participants = std::move(remote_participants_);
-            old_room_handle = std::move(room_handle_);
-            old_e2ee_manager = std::move(e2ee_manager_);
-            old_text_readers = std::move(text_stream_readers_);
-            old_byte_readers = std::move(byte_stream_readers_);
-          }
-
-          // Drain in-flight RPC invocations before destroying the local
-          // participant's FFI handle. Mirrors the ordering in disconnect();
-          // without this, a listener-thread RPC handler can race with handle
-          // disposal and send to a dead handle → INVALID_HANDLE → terminate.
-          if (old_local_participant) {
-            old_local_participant->shutdown();
-          }
-
-          // Remove listener outside lock
-          if (listener_to_remove != 0) {
-            FfiClient::instance().removeListener(listener_to_remove);
-          }
-
-          if (old_local_participant) {
-            old_local_participant->shutdown();
-          }
-
-          // old_* state is destroyed here when going out of scope
+          (void)teardown(false, DisconnectReason::Unknown, false);
 
           const RoomEosEvent ev;
           if (delegate_snapshot) {

src/tests/unit/test_room.cpp

@@ -17,13 +17,44 @@
 #include <gtest/gtest.h>
 #include <livekit/livekit.h>
 
+#include <atomic>
 #include <chrono>
 #include <future>
 #include <string>
 
 #include "ffi.pb.h"
+#include "ffi_client.h"
 #include "room_proto_converter.h"
 
+namespace livekit {
+
+struct RoomTestAccess {
+  static void installConnectedListener(Room& room, std::atomic<int>& 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<std::mutex> guard(room.lock_);
+    room.connection_state_ = ConnectionState::Connected;
+    room.room_handle_ = std::make_shared<FfiHandle>();
+    room.listener_id_ = listener_id;
+    room.teardown_started_ = false;
+  }
+
+  static bool hasRoomHandle(const Room& room) {
+    const std::scoped_lock<std::mutex> guard(room.lock_);
+    return static_cast<bool>(room.room_handle_);
+  }
+
+  static int listenerId(const Room& room) {
+    const std::scoped_lock<std::mutex> guard(room.lock_);
+    return room.listener_id_;
+  }
+};
+
+} // namespace livekit
+
 namespace livekit::test {
 
 class RoomTest : public ::testing::Test {
@@ -33,6 +64,54 @@ class RoomTest : public ::testing::Test {
   void TearDown() override { livekit::shutdown(); }
 };
 
+namespace {
+
+class UnitDisconnectTrackingDelegate : public RoomDelegate {
+public:
+  void onDisconnected(Room&, const DisconnectedEvent& event) override {
+    ++count;
+    reason = event.reason;
+  }
+
+  int count = 0;
+  DisconnectReason reason = DisconnectReason::Unknown;
+};
+
+void emitFfiEvent(const proto::FfiEvent& event) {
+  std::string bytes;
+  ASSERT_TRUE(event.SerializeToString(&bytes));
+  ffiEventCallback(reinterpret_cast<const std::uint8_t*>(bytes.data()), bytes.size());
+}
+
+} // namespace
+
+TEST_F(RoomTest, ServerDisconnectTearsDownRoomAndRemovesListener) {
+  Room room;
+  UnitDisconnectTrackingDelegate delegate;
+  std::atomic<int> listener_calls{0};
+  room.setDelegate(&delegate);
+  RoomTestAccess::installConnectedListener(room, listener_calls);
+
+  proto::FfiEvent event;
+  auto* room_event = event.mutable_room_event();
+  room_event->set_room_handle(0);
+  room_event->mutable_disconnected()->set_reason(proto::ROOM_DELETED);
+
+  emitFfiEvent(event);
+
+  EXPECT_EQ(listener_calls.load(std::memory_order_relaxed), 1);
+  EXPECT_EQ(room.connectionState(), ConnectionState::Disconnected);
+  EXPECT_FALSE(RoomTestAccess::hasRoomHandle(room));
+  EXPECT_EQ(RoomTestAccess::listenerId(room), 0);
+  EXPECT_EQ(delegate.count, 1);
+  EXPECT_EQ(delegate.reason, DisconnectReason::RoomDeleted);
+
+  emitFfiEvent(event);
+  EXPECT_EQ(listener_calls.load(std::memory_order_relaxed), 1) << "server disconnect must unregister the Room listener";
+  EXPECT_EQ(delegate.count, 1) << "server disconnect must notify the delegate exactly once";
+  EXPECT_FALSE(room.disconnect()) << "disconnect after server teardown must be a no-op";
+}
+
 TEST_F(RoomTest, ConnectWithoutInitialize) {
   // Test fixture initializes by default, do this to emulate lack of initialization
   livekit::shutdown();