Bugfix: Server-initiated room disconnect shutdown fix

.token_helpers/set_data_track_test_tokens.bash

@@ -21,6 +21,7 @@
 #   eval "$(bash .token_helpers/set_data_track_test_tokens.bash)"
 #
 # Exports:
+#   LIVEKIT_ROOM
 #   LIVEKIT_TOKEN_A
 #   LIVEKIT_TOKEN_B
 #   LIVEKIT_URL=ws://localhost:7880
@@ -107,12 +108,14 @@ LIVEKIT_TOKEN_A="$(_create_token "$LIVEKIT_IDENTITY_A")"
 LIVEKIT_TOKEN_B="$(_create_token "$LIVEKIT_IDENTITY_B")"
 
 _apply() {
+  export LIVEKIT_ROOM
   export LIVEKIT_TOKEN_A
   export LIVEKIT_TOKEN_B
   export LIVEKIT_URL
 }
 
 _emit_eval() {
+  printf 'export LIVEKIT_ROOM=%q\n' "$LIVEKIT_ROOM"
   printf 'export LIVEKIT_TOKEN_A=%q\n' "$LIVEKIT_TOKEN_A"
   printf 'export LIVEKIT_TOKEN_B=%q\n' "$LIVEKIT_TOKEN_B"
   printf 'export LIVEKIT_URL=%q\n' "$LIVEKIT_URL"

include/livekit/room.h

@@ -330,7 +330,7 @@ class LIVEKIT_API Room {
   void removeOnDataFrameCallback(DataFrameCallbackId id);
 
 private:
-  friend class RoomCallbackTest;
+  friend struct RoomTestAccess;
 
   mutable std::mutex lock_;
   ConnectionState connection_state_ = ConnectionState::Disconnected;
@@ -355,5 +355,8 @@ class LIVEKIT_API Room {
   int listener_id_{0};
 
   void onEvent(const proto::FfiEvent& event);
+
+  // Shared shutdown path for explicit disconnect, server disconnect, EOS, and destruction.
+  bool shutdown(bool disconnect_ffi, DisconnectReason reason, bool notify_delegate);
 };
 } // namespace livekit

src/room.cpp

@@ -212,7 +212,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 shutdown(true, reason, true);
+}
 
+bool Room::shutdown(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;
@@ -221,61 +224,90 @@ bool Room::disconnect(DisconnectReason reason) {
   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;
+  bool claimed_disconnect = false;
 
   {
     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();
+    // Return false for a no-op so callers can tell whether this call claimed the
+    // room state and performed cleanup. Matches disconnect()'s documented contract.
+    if (!has_room_state) {
       return false;
     }
-    handle = room_handle_;
+    // The state transition determines which racing path owns the FFI request
+    // and delegate notification. Remaining room state is still claimed here so
+    // EOS or destruction can finish local cleanup after a server disconnect.
+    claimed_disconnect = connection_state_ != ConnectionState::Disconnected;
+    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 shutdown_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 shutdown: local participant shutdown failed: {}", e.what());
+      shutdown_ok = false;
+    } catch (...) {
+      LK_LOG_ERROR("Room shutdown: local participant shutdown failed: unknown exception");
+      shutdown_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 && claimed_disconnect && 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 shutdown: FFI disconnect failed (continuing local shutdown): {}", e.what());
+      shutdown_ok = false;
+    } catch (...) {
+      LK_LOG_ERROR("Room shutdown: FFI disconnect failed (continuing local shutdown): unknown exception");
+      shutdown_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 shutdown: subscription shutdown failed: {}", e.what());
+      shutdown_ok = false;
+    } catch (...) {
+      LK_LOG_ERROR("Room shutdown: subscription shutdown failed: unknown exception");
+      shutdown_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 shutdown: listener removal failed: {}", e.what());
+      shutdown_ok = false;
+    } catch (...) {
+      LK_LOG_ERROR("Room shutdown: listener removal failed: unknown exception");
+      shutdown_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 && claimed_disconnect && delegate_snapshot) {
     DisconnectedEvent ev;
     ev.reason = reason;
     try {
@@ -287,9 +319,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 claimed_disconnect && shutdown_ok;
 }
 
 RoomInfoData Room::roomInfo() const {
@@ -1168,20 +1198,18 @@ 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;
+          bool should_notify = false;
           {
             const std::scoped_lock<std::mutex> guard(lock_);
-            already_disconnected = (connection_state_ == ConnectionState::Disconnected);
+            // Local shutdown marks the state before awaiting the FFI response
+            // and notifies the delegate itself. Suppress that duplicate while
+            // passing server-initiated disconnects through unchanged.
+            should_notify = connection_state_ != ConnectionState::Disconnected;
             connection_state_ = ConnectionState::Disconnected;
           }
-          if (already_disconnected) {
-            break;
-          }
-          DisconnectedEvent ev;
-          ev.reason = toDisconnectReason(re.disconnected().reason());
-          if (delegate_snapshot) {
+          if (should_notify && delegate_snapshot) {
+            DisconnectedEvent ev;
+            ev.reason = toDisconnectReason(re.disconnected().reason());
             delegate_snapshot->onDisconnected(*this, ev);
           }
           break;
@@ -1208,56 +1236,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)shutdown(false, DisconnectReason::Unknown, false);
 
           const RoomEosEvent ev;
           if (delegate_snapshot) {

src/tests/common/ffi_utils.h

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2026 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <string>
+
+#include "ffi.pb.h"
+#include "ffi_client.h"
+
+namespace livekit::test {
+
+/// Serializes and dispatches a synthetic FFI event through the real callback entry point.
+/// Defined in this header for use across different tests.
+inline 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 livekit::test