Cleanup, shutdown local participants before

Author: alan-george-lk

.github/workflows/tests.yml

@@ -261,11 +261,7 @@ jobs:
         timeout-minutes: 5
         shell: bash
         env:
-          # Bump livekit_ffi to debug so handle-table errors (the exact Rust
-          # message that produces our INVALID_HANDLE → terminate path) land
-          # in the CI log. Keep livekit at info to avoid drowning out the
-          # interesting lines.
-          RUST_LOG: "metrics=debug,livekit_ffi=debug,livekit=info"
+          RUST_LOG: "metrics=debug"
         run: |
           set -euo pipefail
           source .token_helpers/set_data_track_test_tokens.bash

src/ffi_client.cpp

@@ -202,10 +202,6 @@ proto::FfiResponse FfiClient::sendRequest(const proto::FfiRequest& request) cons
   const FfiHandleId handle =
       livekit_ffi_request(reinterpret_cast<const uint8_t*>(bytes.data()), bytes.size(), &resp_ptr, &resp_len);
   if (handle == INVALID_HANDLE) {
-    LK_LOG_ERROR(
-        "FfiClient::sendRequest: livekit_ffi_request returned INVALID_HANDLE; "
-        "request.message_case={}, bytes_len={}",
-        static_cast<int>(request.message_case()), bytes.size());
     throw std::runtime_error("failed to send request, received an invalid handle");
   }
 

src/ffi_handle.cpp

@@ -17,7 +17,6 @@
 #include "livekit/ffi_handle.h"
 
 #include "livekit_ffi.h"
-#include "lk_log.h"
 
 namespace livekit {
 
@@ -36,7 +35,6 @@ FfiHandle& FfiHandle::operator=(FfiHandle&& other) noexcept {
 
 void FfiHandle::reset(uintptr_t new_handle) noexcept {
   if (handle_) {
-    LK_LOG_INFO("FfiHandle::reset: dropping handle {} (replacement={})", handle_, new_handle);
     livekit_ffi_drop_handle(handle_);
   }
   handle_ = new_handle;

src/local_participant.cpp

@@ -370,17 +370,13 @@ void LocalParticipant::shutdown() {
 void LocalParticipant::handleRpcMethodInvocation(uint64_t invocation_id, const std::string& method,
                                                  const std::string& request_id, const std::string& caller_identity,
                                                  const std::string& payload, double response_timeout_sec) {
-  LK_LOG_DEBUG("LocalParticipant::handleRpcMethodInvocation: entry (handle={}, invocation_id={}, method={})",
-               ffiHandleId(), invocation_id, method);
   // Capture shared state so it outlives LocalParticipant if needed
   auto state = rpc_state_;
 
   // Track this invocation and check if we're shutting down
   {
     const std::scoped_lock<std::mutex> lock(state->mutex);
     if (state->shutting_down) {
-      LK_LOG_DEBUG("LocalParticipant::handleRpcMethodInvocation: shutting_down, skipping (invocation_id={})",
-                   invocation_id);
       // Already shutting down, don't process new invocations
       return;
     }
@@ -427,19 +423,14 @@ void LocalParticipant::handleRpcMethodInvocation(uint64_t invocation_id, const s
   {
     const std::scoped_lock<std::mutex> lock(state->mutex);
     if (state->shutting_down) {
-      LK_LOG_DEBUG(
-          "LocalParticipant::handleRpcMethodInvocation: shutdown during handler, skipping response "
-          "(invocation_id={})",
-          invocation_id);
       // Shutdown started, don't send response - handle may be invalid
       return;
     }
   }
 
   FfiRequest req;
   auto* msg = req.mutable_rpc_method_invocation_response();
-  const auto handle_id_at_send = ffiHandleId();
-  msg->set_local_participant_handle(handle_id_at_send);
+  msg->set_local_participant_handle(ffiHandleId());
   msg->set_invocation_id(invocation_id);
   if (response_error.has_value()) {
     auto* err_proto = msg->mutable_error();
@@ -448,8 +439,6 @@ void LocalParticipant::handleRpcMethodInvocation(uint64_t invocation_id, const s
   if (response_payload.has_value()) {
     msg->set_payload(*response_payload);
   }
-  LK_LOG_DEBUG("LocalParticipant::handleRpcMethodInvocation: sending response (handle={}, invocation_id={})",
-               handle_id_at_send, invocation_id);
   FfiClient::instance().sendRequest(req);
 }
 

src/room.cpp

@@ -76,42 +76,17 @@ void readyForRoomEvent(std::uint64_t room_handle) {
 Room::Room() : subscription_thread_dispatcher_(std::make_unique<SubscriptionThreadDispatcher>()) {}
 
 Room::~Room() {
-  LK_LOG_INFO("Room::~Room: entry (this={})", static_cast<const void*>(this));
-  // Issue a graceful disconnect so the server sees us leave instead of
-  // timing out (RAII expectation; see issue #118). disconnect() does the
-  // full teardown including subscription threads, listener, and local
-  // participant, so the destructor only needs to handle the
-  // already-disconnected path.
+  // disconnect() is used for all tear down cases: it handles the
+  // already-disconnected case (returns false, no-op), the partial/Reconnecting
+  // case, and the FFI-failure case (local teardown still runs). Nothing else
+  // needs to live in the destructor.
   try {
-    disconnect();
+    (void)disconnect(); // Don't need return value
   } catch (const std::exception& e) {
     LK_LOG_ERROR("Room::~Room: graceful disconnect failed: {}", e.what());
   } catch (...) {
     LK_LOG_ERROR("Room::~Room: graceful disconnect failed: unknown exception");
   }
-
-  // Defensive: if disconnect() bailed early (e.g. never connected), still
-  // tear down any state that may have leaked.
-  if (subscription_thread_dispatcher_) {
-    subscription_thread_dispatcher_->stopAll();
-  }
-
-  int listener_to_remove = 0;
-  std::unique_ptr<LocalParticipant> local_participant_to_cleanup;
-  {
-    const std::scoped_lock<std::mutex> g(lock_);
-    listener_to_remove = listener_id_;
-    listener_id_ = 0;
-    local_participant_to_cleanup = std::move(local_participant_);
-  }
-
-  if (local_participant_to_cleanup) {
-    local_participant_to_cleanup->shutdown();
-  }
-
-  if (listener_to_remove != 0) {
-    FfiClient::instance().removeListener(listener_to_remove);
-  }
 }
 
 void Room::setDelegate(RoomDelegate* delegate) {
@@ -246,66 +221,79 @@ bool Room::Connect(const std::string& url, const std::string& token, const RoomO
 
 bool Room::disconnect(DisconnectReason reason) {
   TRACE_EVENT0("livekit", "Room::disconnect");
-  LK_LOG_INFO("Room::disconnect: entry (this={}, reason={})", static_cast<const void*>(this), static_cast<int>(reason));
 
-  // Hold onto this in case the
-  auto prev_connection_state = connection_state_;
+  // Canonical teardown path. Move all owned state out under the lock, then
+  // operate on it outside the lock. The destructor (and any caller) gets
+  // the same behavior: once this returns, the Room is fully torn down.
+  //
+  // Return value:
+  //   true  - we owned live state and tore it down (FFI disconnect succeeded)
+  //   false - either already disconnected (no-op) or FFI disconnect failed.
+  //           In both false cases local-side teardown still completed.
 
   std::shared_ptr<FfiHandle> handle;
   RoomDelegate* delegate_snapshot = nullptr;
+  std::unique_ptr<LocalParticipant> local_participant_to_cleanup;
+  std::unordered_map<std::string, std::shared_ptr<RemoteParticipant>> remote_participants_to_clear;
+  std::unique_ptr<E2EEManager> e2ee_manager_to_clear;
+  std::unordered_map<std::string, std::shared_ptr<TextStreamReader>> text_stream_readers_to_clear;
+  std::unordered_map<std::string, std::shared_ptr<ByteStreamReader>> byte_stream_readers_to_clear;
+  int listener_to_remove = 0;
+
   {
     const std::scoped_lock<std::mutex> g(lock_);
     if (connection_state_ == ConnectionState::Disconnected) {
-      LK_LOG_INFO("Room::disconnect: already disconnected, returning false (this={})", static_cast<const void*>(this));
+      // Already torn down (or never connected). Nothing to do.
       return false;
     }
     handle = 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(), which also flips state before sending the request.
     connection_state_ = ConnectionState::Disconnected;
   }
 
-  // Tell the FFI to close the room and wait for the callback. Catch the
-  // exception so we still run teardown below; the caller learns about the
-  // failure via the returned bool / logs.
+  // Drain in-flight RPC handlers BEFORE telling Rust to tear down the room.
+  // Mirrors client-sdk-python's Room.disconnect() ordering: once the FFI
+  // dispatches the Disconnect, Rust starts invalidating participant handles
+  // in its table, and any listener-thread RPC handler still mid-flight
+  // would race with that invalidation and send to a dead handle →
+  // INVALID_HANDLE → terminate.
+  if (local_participant_to_cleanup) {
+    local_participant_to_cleanup->shutdown();
+  }
+
+  // Tell the FFI to close the room and wait for the callback. If this fails
+  // we still complete local-side teardown below — releasing the listener,
+  // dropping handles, and notifying the delegate — so the Room is fully
+  // cleaned up regardless of whether the FFI round-trip succeeded.
   bool ffi_ok = true;
   if (handle) {
     try {
       FfiClient::instance().disconnectAsync(handle->get(), reason).get();
     } catch (const std::exception& e) {
-      LK_LOG_ERROR("Room::disconnect: FFI disconnect failed: {}", e.what());
+      LK_LOG_ERROR("Room::disconnect: FFI disconnect failed (continuing local teardown): {}", e.what());
       ffi_ok = false;
     }
   }
 
-  // Stop dispatcher first so no track callbacks fire mid-teardown.
+  // Stop dispatcher so no track callbacks fire mid-teardown.
   if (subscription_thread_dispatcher_) {
     subscription_thread_dispatcher_->stopAll();
   }
 
-  int listener_to_remove = 0;
-  std::unique_ptr<LocalParticipant> local_participant_to_cleanup;
-  {
-    const std::scoped_lock<std::mutex> g(lock_);
-    listener_to_remove = listener_id_;
-    listener_id_ = 0;
-    local_participant_to_cleanup = std::move(local_participant_);
-    remote_participants_.clear();
-    room_handle_.reset();
-    e2ee_manager_.reset();
-    text_stream_readers_.clear();
-    byte_stream_readers_.clear();
-  }
-
-  // Shut down local participant (unregisters RPC handlers, etc.) before
-  // removing the listener, so in-flight RPC responses don't reach a
-  // destroyed handle.
-  if (local_participant_to_cleanup) {
-    local_participant_to_cleanup->shutdown();
-  }
-
   if (listener_to_remove != 0) {
     FfiClient::instance().removeListener(listener_to_remove);
   }
@@ -323,6 +311,8 @@ bool Room::disconnect(DisconnectReason reason) {
     }
   }
 
+  // Moved-out state (local participant, remote participants, e2ee manager,
+  // stream readers) destructs here, releasing FFI handles.
   return ffi_ok;
 }
 
@@ -1231,8 +1221,6 @@ void Room::onEvent(const FfiEvent& event) {
           break;
         }
         case proto::RoomEvent::kDisconnected: {
-          LK_LOG_INFO("Room::onFfiEvent: kDisconnected received (this={}, reason={})", static_cast<const void*>(this),
-                      static_cast<int>(re.disconnected().reason()));
           // 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;
@@ -1266,7 +1254,6 @@ void Room::onEvent(const FfiEvent& event) {
           break;
         }
         case proto::RoomEvent::kEos: {
-          LK_LOG_INFO("Room::onFfiEvent: kEos received (this={})", static_cast<const void*>(this));
           if (subscription_thread_dispatcher_) {
             subscription_thread_dispatcher_->stopAll();
           }
@@ -1304,8 +1291,6 @@ void Room::onEvent(const FfiEvent& event) {
           // 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) {
-            LK_LOG_INFO("Room::onFfiEvent: kEos shutting down local participant (handle={})",
-                        old_local_participant->ffiHandleId());
             old_local_participant->shutdown();
           }
 

src/tests/integration/test_room.cpp

@@ -96,19 +96,19 @@ TEST_F(RoomTest, UserDisconnect) {
   ASSERT_EQ(room.connectionState(), ConnectionState::Connected);
   ASSERT_NE(room.localParticipant(), nullptr);
 
-  EXPECT_TRUE(room.disconnect()) << "disconnect should report success on a connected room";
+  EXPECT_NO_THROW(room.disconnect()) << "disconnect should not throw on a connected room";
   EXPECT_EQ(room.connectionState(), ConnectionState::Disconnected);
   EXPECT_EQ(room.localParticipant(), nullptr) << "local participant should be cleared after disconnect";
   EXPECT_EQ(delegate.count.load(), 1) << "onDisconnected should fire exactly once";
   EXPECT_EQ(delegate.last_reason, DisconnectReason::ClientInitiated);
 
   // Calling again on an already-disconnected room is a no-op
-  EXPECT_FALSE(room.disconnect()) << "second disconnect should report no-op";
+  EXPECT_NO_THROW(room.disconnect()) << "second disconnect should not throw on an already-disconnected room";
   EXPECT_EQ(delegate.count.load(), 1) << "delegate must not double-fire";
 }
 
 // Case: Room goes out of scope while still connected
-TEST_F(RoomTest, RoomDestructorDisconnect) {
+TEST_F(RoomTest, DestructorDisconnect) {
   std::unique_ptr<Room> room = std::make_unique<Room>();
 
   DisconnectTrackingDelegate delegate;
@@ -117,9 +117,8 @@ TEST_F(RoomTest, RoomDestructorDisconnect) {
   ASSERT_TRUE(room->connect(server_url_, token_, options));
   ASSERT_EQ(room->connectionState(), ConnectionState::Connected);
 
-  room.reset();
+  room.reset(); // invokes destructor which calls disconnect()
 
-  // Let room go out of scope while still connected
   EXPECT_EQ(delegate.count.load(), 1) << "destructor should fire onDisconnected exactly once";
   EXPECT_EQ(delegate.last_reason, DisconnectReason::ClientInitiated);
 }