diff --git a/.token_helpers/set_data_track_test_tokens.bash b/.token_helpers/set_data_track_test_tokens.bash index 3b8d712c..637a5a1c 100755 --- a/.token_helpers/set_data_track_test_tokens.bash +++ b/.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" diff --git a/include/livekit/room.h b/include/livekit/room.h index bf4bf68d..7959ebf3 100644 --- a/include/livekit/room.h +++ b/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,10 @@ class LIVEKIT_API Room { int listener_id_{0}; void onEvent(const proto::FfiEvent& event); + + // Tracks local shutdown independently from connection_state_ to handle all shutdown cases. + bool shutdown_started_{false}; + // Shared shutdown path for explicit disconnect, server disconnect, EOS, and destruction. + bool shutdown(bool disconnect_ffi, DisconnectReason reason, bool notify_delegate); }; } // namespace livekit diff --git a/src/room.cpp b/src/room.cpp index 32dac5eb..e5cebaad 100644 --- a/src/room.cpp +++ b/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"); } + shutdown_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 shutdown(true, reason, true); +} +bool Room::shutdown(bool disconnect_ffi, DisconnectReason reason, bool notify_delegate) { std::shared_ptr handle; RoomDelegate* delegate_snapshot = nullptr; std::shared_ptr local_participant_to_cleanup; @@ -224,15 +228,16 @@ bool Room::disconnect(DisconnectReason reason) { { const std::scoped_lock 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 no-op / in-progress shutdown so callers can tell whether *this* call + // performed cleanup. Matches disconnect()'s documented contract. + if (shutdown_started_ || !has_room_state) { return false; } - handle = room_handle_; + shutdown_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_); @@ -240,42 +245,66 @@ bool Room::disconnect(DisconnectReason reason) { 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 && 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 && delegate_snapshot) { DisconnectedEvent ev; ev.reason = reason; try { @@ -287,9 +316,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 shutdown_ok; } RoomInfoData Room::roomInfo() const { @@ -1168,22 +1195,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 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)shutdown(false, reason, true); break; } case proto::RoomEvent::kReconnecting: { @@ -1208,57 +1221,6 @@ 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 old_local_participant; - std::unordered_map> old_remote_participants; - std::shared_ptr old_room_handle; - std::shared_ptr old_e2ee_manager; - std::unordered_map> old_text_readers; - std::unordered_map> old_byte_readers; - - { - const std::scoped_lock 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 - const RoomEosEvent ev; if (delegate_snapshot) { delegate_snapshot->onRoomEos(*this, ev); diff --git a/src/tests/common/ffi_utils.h b/src/tests/common/ffi_utils.h new file mode 100644 index 00000000..c62d48bb --- /dev/null +++ b/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 + +#include +#include + +#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(bytes.data()), bytes.size()); +} + +} // namespace livekit::test diff --git a/src/tests/integration/test_room.cpp b/src/tests/integration/test_room.cpp index 0e25c901..6190bf88 100644 --- a/src/tests/integration/test_room.cpp +++ b/src/tests/integration/test_room.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -139,12 +140,30 @@ namespace { class DisconnectTrackingDelegate : public RoomDelegate { public: void onDisconnected(Room&, const DisconnectedEvent& ev) override { - ++count; - last_reason = ev.reason; + { + const std::scoped_lock lock(mutex_); + last_reason_ = ev.reason; + } + count.fetch_add(1); + cv_.notify_all(); + } + + bool waitForDisconnect(std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, [this]() { return count.load() > 0; }); + } + + DisconnectReason lastReason() const { + const std::scoped_lock lock(mutex_); + return last_reason_; } std::atomic count{0}; - DisconnectReason last_reason = DisconnectReason::Unknown; + +private: + mutable std::mutex mutex_; + std::condition_variable cv_; + DisconnectReason last_reason_ = DisconnectReason::Unknown; }; class TokenRefreshTrackingDelegate : public RoomDelegate { @@ -221,7 +240,7 @@ TEST_F(RoomTest, UserDisconnect) { EXPECT_EQ(room.connectionState(), ConnectionState::Disconnected); EXPECT_EQ(room.localParticipant().lock(), 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); + EXPECT_EQ(delegate.lastReason(), DisconnectReason::ClientInitiated); // Calling again on an already-disconnected room is a no-op EXPECT_NO_THROW(room.disconnect()) << "second disconnect should not throw on an already-disconnected room"; @@ -243,7 +262,65 @@ TEST_F(RoomTest, DestructorDisconnect) { room.reset(); // invokes destructor which calls disconnect() EXPECT_EQ(delegate.count.load(), 1) << "destructor should fire onDisconnected exactly once"; - EXPECT_EQ(delegate.last_reason, DisconnectReason::ClientInitiated); + EXPECT_EQ(delegate.lastReason(), DisconnectReason::ClientInitiated); +} + +// Case: server deletes the room while the client is connected. +TEST_F(RoomTest, ServerDeletedRoomDisconnectsAndTearsDownLocally) { + ASSERT_TRUE(server_available_) << "LIVEKIT_URL and LIVEKIT_TOKEN_A not set"; + if (server_url_ != kLocalTestLiveKitUrl) { + GTEST_SKIP() << "server-delete integration test requires local livekit-server"; + } + const char* room_name = std::getenv("LIVEKIT_ROOM"); + ASSERT_NE(room_name, nullptr) << "LIVEKIT_ROOM not set"; + ASSERT_NE(std::string(room_name), "") << "LIVEKIT_ROOM must be non-empty"; + + Room room; + DisconnectTrackingDelegate delegate; + room.setDelegate(&delegate); + + RoomOptions options; + ASSERT_TRUE(room.connect(server_url_, token_, options)) << "connect failed"; + ASSERT_EQ(room.connectionState(), ConnectionState::Connected); + ASSERT_NE(room.localParticipant().lock(), nullptr); + + const std::string delete_command = std::string("lk --dev room delete --yes ") + room_name; + ASSERT_EQ(std::system(delete_command.c_str()), 0) << "failed to delete local test room"; + + ASSERT_TRUE(delegate.waitForDisconnect(10s)) << "server room deletion should disconnect the client"; + EXPECT_EQ(room.connectionState(), ConnectionState::Disconnected); + EXPECT_EQ(room.localParticipant().lock(), nullptr) << "local participant should be cleared after server disconnect"; + EXPECT_EQ(delegate.count.load(), 1) << "onDisconnected should fire exactly once"; + EXPECT_EQ(delegate.lastReason(), DisconnectReason::RoomDeleted); + + EXPECT_FALSE(room.disconnect()) << "disconnect after server teardown should be a no-op"; + EXPECT_EQ(delegate.count.load(), 1) << "delegate must not double-fire"; +} + +// Case: same Room instance can connect again after an explicit local disconnect. +TEST_F(RoomTest, ReconnectAfterDisconnect) { + ASSERT_TRUE(server_available_) << "LIVEKIT_URL and LIVEKIT_TOKEN_A not set"; + + Room room; + DisconnectTrackingDelegate delegate; + room.setDelegate(&delegate); + + RoomOptions options; + ASSERT_TRUE(room.connect(server_url_, token_, options)) << "initial connect failed"; + ASSERT_EQ(room.connectionState(), ConnectionState::Connected); + ASSERT_NE(room.localParticipant().lock(), nullptr); + + ASSERT_TRUE(room.disconnect()) << "explicit disconnect should succeed"; + EXPECT_EQ(room.connectionState(), ConnectionState::Disconnected); + EXPECT_EQ(room.localParticipant().lock(), nullptr); + EXPECT_EQ(delegate.count.load(), 1); + + ASSERT_TRUE(room.connect(server_url_, token_, options)) << "reconnect after disconnect should succeed"; + EXPECT_EQ(room.connectionState(), ConnectionState::Connected); + EXPECT_NE(room.localParticipant().lock(), nullptr); + + ASSERT_TRUE(room.disconnect()); + EXPECT_EQ(delegate.count.load(), 2); } // Verifies that participant handles handed out by Room expire once the Room is diff --git a/src/tests/unit/test_ffi_client.cpp b/src/tests/unit/test_ffi_client.cpp index ae17c599..2f41583a 100644 --- a/src/tests/unit/test_ffi_client.cpp +++ b/src/tests/unit/test_ffi_client.cpp @@ -26,6 +26,7 @@ #include #include +#include "../common/ffi_utils.h" #include "ffi.pb.h" #include "ffi_client.h" @@ -54,9 +55,7 @@ void emitEvent() { record->set_target("test"); record->set_message("listener event"); - std::string bytes; - ASSERT_TRUE(event.SerializeToString(&bytes)); - ffiEventCallback(reinterpret_cast(bytes.data()), bytes.size()); + emitFfiEvent(event); } // Minimal stand-in for Room that mirrors its relationship to FfiClient: @@ -408,11 +407,9 @@ TEST_F(FfiClientTest, PanicEvent) { proto::FfiEvent event; event.mutable_panic()->set_message("rust panic"); - std::string bytes; - ASSERT_TRUE(event.SerializeToString(&bytes)); testing::internal::CaptureStderr(); - ffiEventCallback(reinterpret_cast(bytes.data()), bytes.size()); + emitFfiEvent(event); const std::string stderr_output = testing::internal::GetCapturedStderr(); ASSERT_NE(std::signal(SIGTERM, previous_handler), SIG_ERR); diff --git a/src/tests/unit/test_room.cpp b/src/tests/unit/test_room.cpp index f38c2cf6..6224b275 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -17,13 +17,56 @@ #include #include +#include #include #include #include +#include "../common/ffi_utils.h" #include "ffi.pb.h" +#include "ffi_client.h" #include "room_proto_converter.h" +namespace livekit { + +struct RoomTestAccess { + static void installConnectedListener(Room& room, std::atomic& 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 guard(room.lock_); + room.connection_state_ = ConnectionState::Connected; + room.room_handle_ = std::make_shared(); + room.listener_id_ = listener_id; + room.shutdown_started_ = false; + } + + static bool hasRoomHandle(const Room& room) { + const std::scoped_lock guard(room.lock_); + return static_cast(room.room_handle_); + } + + static int listenerId(const Room& room) { + const std::scoped_lock guard(room.lock_); + return room.listener_id_; + } + + static bool shutdownStarted(const Room& room) { + const std::scoped_lock guard(room.lock_); + return room.shutdown_started_; + } + + // Mirrors Room::connect()'s latch clear before FFI work. + static void clearShutdownForReconnect(Room& room) { + const std::scoped_lock guard(room.lock_); + room.shutdown_started_ = false; + } +}; + +} // namespace livekit + namespace livekit::test { class RoomTest : public ::testing::Test { @@ -33,6 +76,21 @@ 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; +}; + +} // namespace + TEST_F(RoomTest, ConnectWithoutInitialize) { // Test fixture initializes by default, do this to emulate lack of initialization livekit::shutdown(); @@ -279,4 +337,31 @@ TEST_F(RoomTest, RemoteParticipantLookupBeforeConnect) { << "Looking up participant before connect should return an empty handle"; } +TEST_F(RoomTest, ServerDisconnectTearsDownRoomAndRemovesListener) { + Room room; + UnitDisconnectTrackingDelegate delegate; + std::atomic 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 shutdown must be a no-op"; +} + } // namespace livekit::test