From 8f5433c3abefc99674e78fcd1adefcb9237a3f9f Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 13 Jul 2026 10:17:39 -0600 Subject: [PATCH 01/13] Initial impl of shutdown repro --- src/tests/CMakeLists.txt | 41 +++ src/tests/manual/server_disconnect_repro.cpp | 322 +++++++++++++++++++ 2 files changed, 363 insertions(+) create mode 100644 src/tests/manual/server_disconnect_repro.cpp diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt index 9583af73..549ca027 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -327,6 +327,47 @@ if(STRESS_TEST_SOURCES) ) endif() +# ============================================================================ +# Manual Reproduction Binaries +# ============================================================================ + +add_executable(server_disconnect_repro + "${CMAKE_CURRENT_SOURCE_DIR}/manual/server_disconnect_repro.cpp" +) +target_link_libraries(server_disconnect_repro PRIVATE livekit) + +if(WIN32) + add_custom_command(TARGET server_disconnect_repro POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ + $ + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "$/livekit_ffi.dll" + $ + COMMENT "Copying DLLs to server disconnect reproducer directory" + ) +elseif(APPLE) + add_custom_command(TARGET server_disconnect_repro POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ + $ + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "$/liblivekit_ffi.dylib" + $ + COMMENT "Copying dylibs to server disconnect reproducer directory" + ) +else() + add_custom_command(TARGET server_disconnect_repro POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ + $ + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "$/liblivekit_ffi.so" + $ + COMMENT "Copying shared libraries to server disconnect reproducer directory" + ) +endif() + # ============================================================================ # Combined test target # ============================================================================ diff --git a/src/tests/manual/server_disconnect_repro.cpp b/src/tests/manual/server_disconnect_repro.cpp new file mode 100644 index 00000000..814a3a39 --- /dev/null +++ b/src/tests/manual/server_disconnect_repro.cpp @@ -0,0 +1,322 @@ +/* + * Copyright 2026 LiveKit, Inc. + * + * 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. + */ + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +using namespace std::chrono_literals; + +constexpr int kFrameWidth = 640; +constexpr int kFrameHeight = 360; +constexpr auto kFrameInterval = 100ms; +constexpr auto kDataInterval = 1s; +constexpr auto kPostRoomDestroyProbe = 5s; +constexpr const char* kPublisherIdentity = "cpp-server-disconnect-repro"; +constexpr const char* kSubscriberIdentity = "cpp-server-disconnect-repro-subscriber"; +constexpr const char* kVideoTrackName = "synthetic-camera"; +constexpr const char* kDataTrackName = "shutdown-repro-data"; + +std::atomic g_exit_requested{false}; +std::atomic g_repro_path_completed{false}; + +void handleSignal(int) { g_exit_requested.store(true, std::memory_order_relaxed); } + +const char* disconnectReasonName(livekit::DisconnectReason reason) { + using livekit::DisconnectReason; + switch (reason) { + case DisconnectReason::ClientInitiated: + return "ClientInitiated"; + case DisconnectReason::DuplicateIdentity: + return "DuplicateIdentity"; + case DisconnectReason::ServerShutdown: + return "ServerShutdown"; + case DisconnectReason::ParticipantRemoved: + return "ParticipantRemoved"; + case DisconnectReason::RoomDeleted: + return "RoomDeleted"; + case DisconnectReason::StateMismatch: + return "StateMismatch"; + case DisconnectReason::JoinFailure: + return "JoinFailure"; + case DisconnectReason::Migration: + return "Migration"; + case DisconnectReason::SignalClose: + return "SignalClose"; + case DisconnectReason::RoomClosed: + return "RoomClosed"; + case DisconnectReason::UserUnavailable: + return "UserUnavailable"; + case DisconnectReason::UserRejected: + return "UserRejected"; + case DisconnectReason::SipTrunkFailure: + return "SipTrunkFailure"; + case DisconnectReason::ConnectionTimeout: + return "ConnectionTimeout"; + case DisconnectReason::MediaFailure: + return "MediaFailure"; + case DisconnectReason::Unknown: + return "Unknown"; + } + return "Unknown"; +} + +const char* connectionStateName(livekit::ConnectionState state) { + using livekit::ConnectionState; + switch (state) { + case ConnectionState::Disconnected: + return "Disconnected"; + case ConnectionState::Connected: + return "Connected"; + case ConnectionState::Reconnecting: + return "Reconnecting"; + } + return "Unknown"; +} + +class ReproRoomDelegate final : public livekit::RoomDelegate { +public: + explicit ReproRoomDelegate(std::string label) : label_(std::move(label)) {} + + void onConnectionStateChanged(livekit::Room&, const livekit::ConnectionStateChangedEvent& event) override { + std::cout << "[" << label_ << " delegate] connection state changed: " << connectionStateName(event.state) << "\n"; + } + + void onDisconnected(livekit::Room&, const livekit::DisconnectedEvent& event) override { + disconnected_.store(true, std::memory_order_relaxed); + std::cout << "[" << label_ << " delegate] disconnected: reason=" << disconnectReasonName(event.reason) << " (" + << static_cast(event.reason) << ")\n"; + std::cout << "[" << label_ << " delegate] waiting for SIGINT; Room remains alive\n"; + } + + void onRoomEos(livekit::Room&, const livekit::RoomEosEvent&) override { + std::cout << "[" << label_ << " delegate] room EOS received\n"; + } + + bool disconnected() const noexcept { return disconnected_.load(std::memory_order_relaxed); } + +private: + std::string label_; + std::atomic disconnected_{false}; +}; + +void fillFrame(livekit::VideoFrame& frame, std::uint32_t frame_index) { + const auto blue = static_cast((frame_index * 7U) % 255U); + const auto green = static_cast((frame_index * 13U) % 255U); + const auto red = static_cast((frame_index * 29U) % 255U); + + std::uint8_t* data = frame.data(); + for (std::size_t i = 0; i < frame.dataSize(); i += 4) { + data[i] = blue; + data[i + 1] = green; + data[i + 2] = red; + data[i + 3] = 255; + } +} + +void publishFrames(const std::shared_ptr& video_source, + const std::shared_ptr& data_track, const std::atomic& publishing) { + livekit::VideoFrame frame = livekit::VideoFrame::create(kFrameWidth, kFrameHeight, livekit::VideoBufferType::BGRA); + const auto started_at = std::chrono::steady_clock::now(); + auto next_frame_at = started_at; + auto next_data_at = started_at; + std::uint32_t frame_index = 0; + std::uint64_t data_index = 0; + bool capture_error_logged = false; + bool data_error_logged = false; + + while (publishing.load(std::memory_order_relaxed)) { + fillFrame(frame, frame_index++); + const auto now = std::chrono::steady_clock::now(); + const auto timestamp = std::chrono::duration_cast(now - started_at).count(); + try { + video_source->captureFrame(frame, static_cast(timestamp)); + } catch (const std::exception& error) { + if (!capture_error_logged) { + std::cerr << "[publisher] video capture failed, continuing repro loop: " << error.what() << "\n"; + capture_error_logged = true; + } + } + + if (now >= next_data_at) { + const std::string message = "server-disconnect-repro frame " + std::to_string(data_index++); + std::vector payload(message.begin(), message.end()); + const auto push_result = data_track->tryPush(std::move(payload)); + if (!push_result && !data_error_logged) { + std::cerr << "[publisher] data push failed, continuing repro loop: " << push_result.error().message << "\n"; + data_error_logged = true; + } + next_data_at = now + kDataInterval; + } + + next_frame_at += kFrameInterval; + std::this_thread::sleep_until(next_frame_at); + } + + std::cout << "[publisher] capture loop stopped\n"; +} + +int run() { + const char* sandbox_id = std::getenv("LIVEKIT_SANDBOX_ID"); + if (sandbox_id == nullptr || std::string(sandbox_id).empty()) { + std::cerr << "LIVEKIT_SANDBOX_ID must be set\n"; + return 1; + } + + auto token_source = livekit::SandboxTokenSource::create(sandbox_id); + livekit::TokenRequestOptions request_options; + request_options.participant_identity = kPublisherIdentity; + + std::cout << "Fetching sandbox credentials...\n"; + auto credentials_result = token_source->fetch(request_options).get(); + if (!credentials_result) { + std::cerr << "Failed to fetch sandbox credentials: " << credentials_result.error().message << "\n"; + return 1; + } + const auto credentials = std::move(credentials_result).value(); + + ReproRoomDelegate publisher_delegate("publisher"); + auto room = std::make_unique(); + room->setDelegate(&publisher_delegate); + + livekit::RoomOptions room_options; + room_options.auto_subscribe = true; + room_options.dynacast = false; + if (!room->connect(credentials.server_url, credentials.participant_token, room_options)) { + std::cerr << "Failed to connect to the sandbox room\n"; + return 1; + } + + auto local_participant = room->localParticipant().lock(); + if (local_participant == nullptr) { + throw std::runtime_error("Local participant unavailable after connect"); + } + + std::cout << "Connected to room '" << room->roomInfo().name << "' as '" << local_participant->identity() << "'" + << "\n"; + + livekit::TokenRequestOptions subscriber_request_options; + subscriber_request_options.room_name = room->roomInfo().name; + subscriber_request_options.participant_identity = kSubscriberIdentity; + + std::cout << "Fetching sandbox credentials for subscriber...\n"; + auto subscriber_credentials_result = token_source->fetch(subscriber_request_options).get(); + if (!subscriber_credentials_result) { + std::cerr << "Failed to fetch subscriber sandbox credentials: " << subscriber_credentials_result.error().message + << "\n"; + return 1; + } + const auto subscriber_credentials = std::move(subscriber_credentials_result).value(); + + ReproRoomDelegate subscriber_delegate("subscriber"); + auto subscriber_room = std::make_unique(); + subscriber_room->setDelegate(&subscriber_delegate); + subscriber_room->setOnVideoFrameCallback(kPublisherIdentity, kVideoTrackName, + [](const livekit::VideoFrame&, std::int64_t) {}); + const auto data_callback_id = subscriber_room->addOnDataFrameCallback( + kPublisherIdentity, kDataTrackName, [](const std::vector&, std::optional) {}); + + if (!subscriber_room->connect(subscriber_credentials.server_url, subscriber_credentials.participant_token, + room_options)) { + std::cerr << "Failed to connect subscriber to the sandbox room\n"; + return 1; + } + std::cout << "Connected subscriber to the same room; discard callbacks are registered (data callback id=" + << data_callback_id << ")\n"; + + auto video_source = std::make_shared(kFrameWidth, kFrameHeight); + auto video_track = livekit::LocalVideoTrack::createLocalVideoTrack(kVideoTrackName, video_source); + livekit::TrackPublishOptions video_options; + video_options.source = livekit::TrackSource::SOURCE_CAMERA; + video_options.simulcast = false; + local_participant->publishTrack(video_track, video_options); + std::cout << "Published synthetic camera track\n"; + + auto data_track_result = local_participant->publishDataTrack(kDataTrackName); + if (!data_track_result) { + throw std::runtime_error("Failed to publish data track: " + data_track_result.error().message); + } + auto data_track = std::move(data_track_result).value(); + std::cout << "Published data track\n"; + std::cout << "Subscriber is receiving and discarding the published video/data frames\n"; + std::cout << "Delete the room remotely, wait for the disconnect callback, then press Ctrl-C\n"; + + std::atomic publishing{true}; + std::thread publisher([&]() { publishFrames(video_source, data_track, publishing); }); + + while (!g_exit_requested.load(std::memory_order_relaxed)) { + std::this_thread::sleep_for(100ms); + } + + std::cout << "SIGINT received; destroying Rooms without an explicit disconnect while publisher is active...\n"; + local_participant.reset(); + room.reset(); + subscriber_room.reset(); + std::cout << "Rooms destroyed; probing for stale FFI callbacks for " << kPostRoomDestroyProbe.count() << "s\n"; + std::this_thread::sleep_for(kPostRoomDestroyProbe); + + publishing.store(false, std::memory_order_relaxed); + if (publisher.joinable()) { + publisher.join(); + } + + std::cout << "Releasing track resources\n"; + data_track.reset(); + video_track.reset(); + video_source.reset(); + g_repro_path_completed.store(true, std::memory_order_relaxed); + std::cout << "Track resources released\n"; + return 0; +} + +} // namespace + +int main() { + std::signal(SIGINT, handleSignal); + livekit::initialize(livekit::LogLevel::Info); + + int exit_code = 0; + try { + exit_code = run(); + } catch (const std::exception& error) { + std::cerr << "Fatal error: " << error.what() << "\n"; + exit_code = 1; + } + + if (g_repro_path_completed.load(std::memory_order_relaxed)) { + std::cout << "Skipping livekit::shutdown() to preserve the bad-shutdown reproduction path\n"; + } else { + livekit::shutdown(); + } + return exit_code; +} From d5dda8cf120551c0487932f465938b4f626b80d9 Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 13 Jul 2026 10:54:42 -0600 Subject: [PATCH 02/13] Use room name for repro --- src/tests/manual/server_disconnect_repro.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/tests/manual/server_disconnect_repro.cpp b/src/tests/manual/server_disconnect_repro.cpp index 814a3a39..a67c4501 100644 --- a/src/tests/manual/server_disconnect_repro.cpp +++ b/src/tests/manual/server_disconnect_repro.cpp @@ -42,6 +42,7 @@ constexpr int kFrameHeight = 360; constexpr auto kFrameInterval = 100ms; constexpr auto kDataInterval = 1s; constexpr auto kPostRoomDestroyProbe = 5s; +constexpr const char* kRoomName = "cpp-test"; constexpr const char* kPublisherIdentity = "cpp-server-disconnect-repro"; constexpr const char* kSubscriberIdentity = "cpp-server-disconnect-repro-subscriber"; constexpr const char* kVideoTrackName = "synthetic-camera"; @@ -195,6 +196,7 @@ int run() { auto token_source = livekit::SandboxTokenSource::create(sandbox_id); livekit::TokenRequestOptions request_options; + request_options.room_name = kRoomName; request_options.participant_identity = kPublisherIdentity; std::cout << "Fetching sandbox credentials...\n"; @@ -226,7 +228,7 @@ int run() { << "\n"; livekit::TokenRequestOptions subscriber_request_options; - subscriber_request_options.room_name = room->roomInfo().name; + subscriber_request_options.room_name = kRoomName; subscriber_request_options.participant_identity = kSubscriberIdentity; std::cout << "Fetching sandbox credentials for subscriber...\n"; From 0b9ea06a86141b7f7bafbdf8057074b474432b79 Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 13 Jul 2026 11:41:05 -0600 Subject: [PATCH 03/13] Implementation of teardown feature for bugfix --- include/livekit/room.h | 6 ++ src/room.cpp | 148 +++++++++++++---------------------- src/tests/unit/test_room.cpp | 79 +++++++++++++++++++ 3 files changed, 140 insertions(+), 93 deletions(-) diff --git a/include/livekit/room.h b/include/livekit/room.h index bf4bf68d..450ec36e 100644 --- a/include/livekit/room.h +++ b/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 diff --git a/src/room.cpp b/src/room.cpp index 32dac5eb..c3e2edc1 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"); } + 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 handle; RoomDelegate* delegate_snapshot = nullptr; std::shared_ptr local_participant_to_cleanup; @@ -224,15 +228,14 @@ 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(); + 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_); @@ -240,42 +243,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 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 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 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 + (void)teardown(false, DisconnectReason::Unknown, false); const RoomEosEvent ev; if (delegate_snapshot) { diff --git a/src/tests/unit/test_room.cpp b/src/tests/unit/test_room.cpp index f38c2cf6..6f08ca73 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -17,13 +17,44 @@ #include #include +#include #include #include #include #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.teardown_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_; + } +}; + +} // 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(bytes.data()), bytes.size()); +} + +} // namespace + +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 teardown must be a no-op"; +} + TEST_F(RoomTest, ConnectWithoutInitialize) { // Test fixture initializes by default, do this to emulate lack of initialization livekit::shutdown(); From 9effbcb060eb399648665d7f6110ae7a8f4aa385 Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 13 Jul 2026 12:29:54 -0600 Subject: [PATCH 04/13] Add integration test --- .../set_data_track_test_tokens.bash | 3 + src/tests/integration/test_room.cpp | 61 +++++++++++++++++-- 2 files changed, 59 insertions(+), 5 deletions(-) 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/src/tests/integration/test_room.cpp b/src/tests/integration/test_room.cpp index 0e25c901..d3e0f29e 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,39 @@ 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"; } // Verifies that participant handles handed out by Room expire once the Room is From 738693bfe39c7f2b1930f8c11c819a893bfa18c7 Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 13 Jul 2026 12:35:26 -0600 Subject: [PATCH 05/13] Remove standalone disconnect repro from branch Co-authored-by: Cursor --- src/tests/CMakeLists.txt | 41 --- src/tests/manual/server_disconnect_repro.cpp | 324 ------------------- 2 files changed, 365 deletions(-) delete mode 100644 src/tests/manual/server_disconnect_repro.cpp diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt index 549ca027..9583af73 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -327,47 +327,6 @@ if(STRESS_TEST_SOURCES) ) endif() -# ============================================================================ -# Manual Reproduction Binaries -# ============================================================================ - -add_executable(server_disconnect_repro - "${CMAKE_CURRENT_SOURCE_DIR}/manual/server_disconnect_repro.cpp" -) -target_link_libraries(server_disconnect_repro PRIVATE livekit) - -if(WIN32) - add_custom_command(TARGET server_disconnect_repro POST_BUILD - COMMAND ${CMAKE_COMMAND} -E copy_if_different - $ - $ - COMMAND ${CMAKE_COMMAND} -E copy_if_different - "$/livekit_ffi.dll" - $ - COMMENT "Copying DLLs to server disconnect reproducer directory" - ) -elseif(APPLE) - add_custom_command(TARGET server_disconnect_repro POST_BUILD - COMMAND ${CMAKE_COMMAND} -E copy_if_different - $ - $ - COMMAND ${CMAKE_COMMAND} -E copy_if_different - "$/liblivekit_ffi.dylib" - $ - COMMENT "Copying dylibs to server disconnect reproducer directory" - ) -else() - add_custom_command(TARGET server_disconnect_repro POST_BUILD - COMMAND ${CMAKE_COMMAND} -E copy_if_different - $ - $ - COMMAND ${CMAKE_COMMAND} -E copy_if_different - "$/liblivekit_ffi.so" - $ - COMMENT "Copying shared libraries to server disconnect reproducer directory" - ) -endif() - # ============================================================================ # Combined test target # ============================================================================ diff --git a/src/tests/manual/server_disconnect_repro.cpp b/src/tests/manual/server_disconnect_repro.cpp deleted file mode 100644 index a67c4501..00000000 --- a/src/tests/manual/server_disconnect_repro.cpp +++ /dev/null @@ -1,324 +0,0 @@ -/* - * Copyright 2026 LiveKit, Inc. - * - * 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. - */ - -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -namespace { - -using namespace std::chrono_literals; - -constexpr int kFrameWidth = 640; -constexpr int kFrameHeight = 360; -constexpr auto kFrameInterval = 100ms; -constexpr auto kDataInterval = 1s; -constexpr auto kPostRoomDestroyProbe = 5s; -constexpr const char* kRoomName = "cpp-test"; -constexpr const char* kPublisherIdentity = "cpp-server-disconnect-repro"; -constexpr const char* kSubscriberIdentity = "cpp-server-disconnect-repro-subscriber"; -constexpr const char* kVideoTrackName = "synthetic-camera"; -constexpr const char* kDataTrackName = "shutdown-repro-data"; - -std::atomic g_exit_requested{false}; -std::atomic g_repro_path_completed{false}; - -void handleSignal(int) { g_exit_requested.store(true, std::memory_order_relaxed); } - -const char* disconnectReasonName(livekit::DisconnectReason reason) { - using livekit::DisconnectReason; - switch (reason) { - case DisconnectReason::ClientInitiated: - return "ClientInitiated"; - case DisconnectReason::DuplicateIdentity: - return "DuplicateIdentity"; - case DisconnectReason::ServerShutdown: - return "ServerShutdown"; - case DisconnectReason::ParticipantRemoved: - return "ParticipantRemoved"; - case DisconnectReason::RoomDeleted: - return "RoomDeleted"; - case DisconnectReason::StateMismatch: - return "StateMismatch"; - case DisconnectReason::JoinFailure: - return "JoinFailure"; - case DisconnectReason::Migration: - return "Migration"; - case DisconnectReason::SignalClose: - return "SignalClose"; - case DisconnectReason::RoomClosed: - return "RoomClosed"; - case DisconnectReason::UserUnavailable: - return "UserUnavailable"; - case DisconnectReason::UserRejected: - return "UserRejected"; - case DisconnectReason::SipTrunkFailure: - return "SipTrunkFailure"; - case DisconnectReason::ConnectionTimeout: - return "ConnectionTimeout"; - case DisconnectReason::MediaFailure: - return "MediaFailure"; - case DisconnectReason::Unknown: - return "Unknown"; - } - return "Unknown"; -} - -const char* connectionStateName(livekit::ConnectionState state) { - using livekit::ConnectionState; - switch (state) { - case ConnectionState::Disconnected: - return "Disconnected"; - case ConnectionState::Connected: - return "Connected"; - case ConnectionState::Reconnecting: - return "Reconnecting"; - } - return "Unknown"; -} - -class ReproRoomDelegate final : public livekit::RoomDelegate { -public: - explicit ReproRoomDelegate(std::string label) : label_(std::move(label)) {} - - void onConnectionStateChanged(livekit::Room&, const livekit::ConnectionStateChangedEvent& event) override { - std::cout << "[" << label_ << " delegate] connection state changed: " << connectionStateName(event.state) << "\n"; - } - - void onDisconnected(livekit::Room&, const livekit::DisconnectedEvent& event) override { - disconnected_.store(true, std::memory_order_relaxed); - std::cout << "[" << label_ << " delegate] disconnected: reason=" << disconnectReasonName(event.reason) << " (" - << static_cast(event.reason) << ")\n"; - std::cout << "[" << label_ << " delegate] waiting for SIGINT; Room remains alive\n"; - } - - void onRoomEos(livekit::Room&, const livekit::RoomEosEvent&) override { - std::cout << "[" << label_ << " delegate] room EOS received\n"; - } - - bool disconnected() const noexcept { return disconnected_.load(std::memory_order_relaxed); } - -private: - std::string label_; - std::atomic disconnected_{false}; -}; - -void fillFrame(livekit::VideoFrame& frame, std::uint32_t frame_index) { - const auto blue = static_cast((frame_index * 7U) % 255U); - const auto green = static_cast((frame_index * 13U) % 255U); - const auto red = static_cast((frame_index * 29U) % 255U); - - std::uint8_t* data = frame.data(); - for (std::size_t i = 0; i < frame.dataSize(); i += 4) { - data[i] = blue; - data[i + 1] = green; - data[i + 2] = red; - data[i + 3] = 255; - } -} - -void publishFrames(const std::shared_ptr& video_source, - const std::shared_ptr& data_track, const std::atomic& publishing) { - livekit::VideoFrame frame = livekit::VideoFrame::create(kFrameWidth, kFrameHeight, livekit::VideoBufferType::BGRA); - const auto started_at = std::chrono::steady_clock::now(); - auto next_frame_at = started_at; - auto next_data_at = started_at; - std::uint32_t frame_index = 0; - std::uint64_t data_index = 0; - bool capture_error_logged = false; - bool data_error_logged = false; - - while (publishing.load(std::memory_order_relaxed)) { - fillFrame(frame, frame_index++); - const auto now = std::chrono::steady_clock::now(); - const auto timestamp = std::chrono::duration_cast(now - started_at).count(); - try { - video_source->captureFrame(frame, static_cast(timestamp)); - } catch (const std::exception& error) { - if (!capture_error_logged) { - std::cerr << "[publisher] video capture failed, continuing repro loop: " << error.what() << "\n"; - capture_error_logged = true; - } - } - - if (now >= next_data_at) { - const std::string message = "server-disconnect-repro frame " + std::to_string(data_index++); - std::vector payload(message.begin(), message.end()); - const auto push_result = data_track->tryPush(std::move(payload)); - if (!push_result && !data_error_logged) { - std::cerr << "[publisher] data push failed, continuing repro loop: " << push_result.error().message << "\n"; - data_error_logged = true; - } - next_data_at = now + kDataInterval; - } - - next_frame_at += kFrameInterval; - std::this_thread::sleep_until(next_frame_at); - } - - std::cout << "[publisher] capture loop stopped\n"; -} - -int run() { - const char* sandbox_id = std::getenv("LIVEKIT_SANDBOX_ID"); - if (sandbox_id == nullptr || std::string(sandbox_id).empty()) { - std::cerr << "LIVEKIT_SANDBOX_ID must be set\n"; - return 1; - } - - auto token_source = livekit::SandboxTokenSource::create(sandbox_id); - livekit::TokenRequestOptions request_options; - request_options.room_name = kRoomName; - request_options.participant_identity = kPublisherIdentity; - - std::cout << "Fetching sandbox credentials...\n"; - auto credentials_result = token_source->fetch(request_options).get(); - if (!credentials_result) { - std::cerr << "Failed to fetch sandbox credentials: " << credentials_result.error().message << "\n"; - return 1; - } - const auto credentials = std::move(credentials_result).value(); - - ReproRoomDelegate publisher_delegate("publisher"); - auto room = std::make_unique(); - room->setDelegate(&publisher_delegate); - - livekit::RoomOptions room_options; - room_options.auto_subscribe = true; - room_options.dynacast = false; - if (!room->connect(credentials.server_url, credentials.participant_token, room_options)) { - std::cerr << "Failed to connect to the sandbox room\n"; - return 1; - } - - auto local_participant = room->localParticipant().lock(); - if (local_participant == nullptr) { - throw std::runtime_error("Local participant unavailable after connect"); - } - - std::cout << "Connected to room '" << room->roomInfo().name << "' as '" << local_participant->identity() << "'" - << "\n"; - - livekit::TokenRequestOptions subscriber_request_options; - subscriber_request_options.room_name = kRoomName; - subscriber_request_options.participant_identity = kSubscriberIdentity; - - std::cout << "Fetching sandbox credentials for subscriber...\n"; - auto subscriber_credentials_result = token_source->fetch(subscriber_request_options).get(); - if (!subscriber_credentials_result) { - std::cerr << "Failed to fetch subscriber sandbox credentials: " << subscriber_credentials_result.error().message - << "\n"; - return 1; - } - const auto subscriber_credentials = std::move(subscriber_credentials_result).value(); - - ReproRoomDelegate subscriber_delegate("subscriber"); - auto subscriber_room = std::make_unique(); - subscriber_room->setDelegate(&subscriber_delegate); - subscriber_room->setOnVideoFrameCallback(kPublisherIdentity, kVideoTrackName, - [](const livekit::VideoFrame&, std::int64_t) {}); - const auto data_callback_id = subscriber_room->addOnDataFrameCallback( - kPublisherIdentity, kDataTrackName, [](const std::vector&, std::optional) {}); - - if (!subscriber_room->connect(subscriber_credentials.server_url, subscriber_credentials.participant_token, - room_options)) { - std::cerr << "Failed to connect subscriber to the sandbox room\n"; - return 1; - } - std::cout << "Connected subscriber to the same room; discard callbacks are registered (data callback id=" - << data_callback_id << ")\n"; - - auto video_source = std::make_shared(kFrameWidth, kFrameHeight); - auto video_track = livekit::LocalVideoTrack::createLocalVideoTrack(kVideoTrackName, video_source); - livekit::TrackPublishOptions video_options; - video_options.source = livekit::TrackSource::SOURCE_CAMERA; - video_options.simulcast = false; - local_participant->publishTrack(video_track, video_options); - std::cout << "Published synthetic camera track\n"; - - auto data_track_result = local_participant->publishDataTrack(kDataTrackName); - if (!data_track_result) { - throw std::runtime_error("Failed to publish data track: " + data_track_result.error().message); - } - auto data_track = std::move(data_track_result).value(); - std::cout << "Published data track\n"; - std::cout << "Subscriber is receiving and discarding the published video/data frames\n"; - std::cout << "Delete the room remotely, wait for the disconnect callback, then press Ctrl-C\n"; - - std::atomic publishing{true}; - std::thread publisher([&]() { publishFrames(video_source, data_track, publishing); }); - - while (!g_exit_requested.load(std::memory_order_relaxed)) { - std::this_thread::sleep_for(100ms); - } - - std::cout << "SIGINT received; destroying Rooms without an explicit disconnect while publisher is active...\n"; - local_participant.reset(); - room.reset(); - subscriber_room.reset(); - std::cout << "Rooms destroyed; probing for stale FFI callbacks for " << kPostRoomDestroyProbe.count() << "s\n"; - std::this_thread::sleep_for(kPostRoomDestroyProbe); - - publishing.store(false, std::memory_order_relaxed); - if (publisher.joinable()) { - publisher.join(); - } - - std::cout << "Releasing track resources\n"; - data_track.reset(); - video_track.reset(); - video_source.reset(); - g_repro_path_completed.store(true, std::memory_order_relaxed); - std::cout << "Track resources released\n"; - return 0; -} - -} // namespace - -int main() { - std::signal(SIGINT, handleSignal); - livekit::initialize(livekit::LogLevel::Info); - - int exit_code = 0; - try { - exit_code = run(); - } catch (const std::exception& error) { - std::cerr << "Fatal error: " << error.what() << "\n"; - exit_code = 1; - } - - if (g_repro_path_completed.load(std::memory_order_relaxed)) { - std::cout << "Skipping livekit::shutdown() to preserve the bad-shutdown reproduction path\n"; - } else { - livekit::shutdown(); - } - return exit_code; -} From 0672226a34edf52c21f6fffb7c9660e4c527f61e Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 13 Jul 2026 16:05:03 -0600 Subject: [PATCH 06/13] Cleanup --- include/livekit/room.h | 1 - 1 file changed, 1 deletion(-) diff --git a/include/livekit/room.h b/include/livekit/room.h index 450ec36e..150e9388 100644 --- a/include/livekit/room.h +++ b/include/livekit/room.h @@ -330,7 +330,6 @@ class LIVEKIT_API Room { void removeOnDataFrameCallback(DataFrameCallbackId id); private: - friend class RoomCallbackTest; friend struct RoomTestAccess; mutable std::mutex lock_; From 7810fa30bac1218bf32d533043941e73b017dc51 Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 13 Jul 2026 16:35:32 -0600 Subject: [PATCH 07/13] Unify emitFfiEvent into standalone header --- src/tests/common/ffi_utils.h | 37 ++++++++++++++++++++++++++++++ src/tests/unit/test_ffi_client.cpp | 9 +++----- src/tests/unit/test_room.cpp | 7 +----- 3 files changed, 41 insertions(+), 12 deletions(-) create mode 100644 src/tests/common/ffi_utils.h 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/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 6f08ca73..327a7b62 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -22,6 +22,7 @@ #include #include +#include "../common/ffi_utils.h" #include "ffi.pb.h" #include "ffi_client.h" #include "room_proto_converter.h" @@ -77,12 +78,6 @@ class UnitDisconnectTrackingDelegate : public RoomDelegate { DisconnectReason reason = DisconnectReason::Unknown; }; -void emitFfiEvent(const proto::FfiEvent& event) { - std::string bytes; - ASSERT_TRUE(event.SerializeToString(&bytes)); - ffiEventCallback(reinterpret_cast(bytes.data()), bytes.size()); -} - } // namespace TEST_F(RoomTest, ServerDisconnectTearsDownRoomAndRemovesListener) { From 83c403b34f7ffbf045831e905790fb9909588281 Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 13 Jul 2026 21:56:38 -0600 Subject: [PATCH 08/13] Address review: rename Room teardown to shutdown, add reconnect tests Align the shared cleanup helper with LocalParticipant/FfiClient naming, keep disconnect() returning false for no-ops per its documented contract, and cover Room reuse after server-initiated shutdown. Co-authored-by: Cursor --- include/livekit/room.h | 8 ++--- src/room.cpp | 52 +++++++++++++++-------------- src/tests/integration/test_room.cpp | 26 +++++++++++++++ src/tests/unit/test_room.cpp | 47 ++++++++++++++++++++++++-- 4 files changed, 102 insertions(+), 31 deletions(-) diff --git a/include/livekit/room.h b/include/livekit/room.h index 150e9388..7959ebf3 100644 --- a/include/livekit/room.h +++ b/include/livekit/room.h @@ -356,9 +356,9 @@ class LIVEKIT_API Room { 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); + // 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 c3e2edc1..5479712b 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -102,7 +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; + shutdown_started_ = false; connection_state_ = ConnectionState::Reconnecting; } @@ -213,10 +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); + return shutdown(true, reason, true); } -bool Room::teardown(bool disconnect_ffi, DisconnectReason reason, bool notify_delegate) { +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; @@ -230,10 +230,12 @@ bool Room::teardown(bool disconnect_ffi, DisconnectReason reason, bool notify_de const std::scoped_lock g(lock_); 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 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; } - teardown_started_ = true; + shutdown_started_ = true; handle = std::move(room_handle_); delegate_snapshot = delegate_; local_participant_to_cleanup = std::move(local_participant_); @@ -246,16 +248,16 @@ bool Room::teardown(bool disconnect_ffi, DisconnectReason reason, bool notify_de connection_state_ = ConnectionState::Disconnected; } - bool teardown_ok = true; + bool shutdown_ok = true; if (local_participant_to_cleanup) { 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; + LK_LOG_ERROR("Room shutdown: local participant shutdown failed: {}", e.what()); + shutdown_ok = false; } catch (...) { - LK_LOG_ERROR("Room teardown: local participant shutdown failed: unknown exception"); - teardown_ok = false; + LK_LOG_ERROR("Room shutdown: local participant shutdown failed: unknown exception"); + shutdown_ok = false; } } @@ -263,11 +265,11 @@ bool Room::teardown(bool disconnect_ffi, DisconnectReason reason, bool notify_de try { FfiClient::instance().disconnectAsync(handle->get(), reason).get(); } catch (const std::exception& e) { - LK_LOG_ERROR("Room teardown: FFI disconnect failed (continuing local teardown): {}", e.what()); - teardown_ok = false; + LK_LOG_ERROR("Room shutdown: FFI disconnect failed (continuing local shutdown): {}", e.what()); + shutdown_ok = false; } catch (...) { - LK_LOG_ERROR("Room teardown: FFI disconnect failed (continuing local teardown): unknown exception"); - teardown_ok = false; + LK_LOG_ERROR("Room shutdown: FFI disconnect failed (continuing local shutdown): unknown exception"); + shutdown_ok = false; } } @@ -275,11 +277,11 @@ bool Room::teardown(bool disconnect_ffi, DisconnectReason reason, bool notify_de try { subscription_thread_dispatcher_->stopAll(); } catch (const std::exception& e) { - LK_LOG_ERROR("Room teardown: subscription shutdown failed: {}", e.what()); - teardown_ok = false; + LK_LOG_ERROR("Room shutdown: subscription shutdown failed: {}", e.what()); + shutdown_ok = false; } catch (...) { - LK_LOG_ERROR("Room teardown: subscription shutdown failed: unknown exception"); - teardown_ok = false; + LK_LOG_ERROR("Room shutdown: subscription shutdown failed: unknown exception"); + shutdown_ok = false; } } @@ -287,11 +289,11 @@ bool Room::teardown(bool disconnect_ffi, DisconnectReason reason, bool notify_de 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; + LK_LOG_ERROR("Room shutdown: listener removal failed: {}", e.what()); + shutdown_ok = false; } catch (...) { - LK_LOG_ERROR("Room teardown: listener removal failed: unknown exception"); - teardown_ok = false; + LK_LOG_ERROR("Room shutdown: listener removal failed: unknown exception"); + shutdown_ok = false; } } @@ -314,7 +316,7 @@ bool Room::teardown(bool disconnect_ffi, DisconnectReason reason, bool notify_de } } - return teardown_ok; + return shutdown_ok; } RoomInfoData Room::roomInfo() const { @@ -1194,7 +1196,7 @@ void Room::onEvent(const FfiEvent& event) { } case proto::RoomEvent::kDisconnected: { const auto reason = toDisconnectReason(re.disconnected().reason()); - (void)teardown(false, reason, true); + (void)shutdown(false, reason, true); break; } case proto::RoomEvent::kReconnecting: { @@ -1219,7 +1221,7 @@ void Room::onEvent(const FfiEvent& event) { break; } case proto::RoomEvent::kEos: { - (void)teardown(false, DisconnectReason::Unknown, false); + (void)shutdown(false, DisconnectReason::Unknown, false); const RoomEosEvent ev; if (delegate_snapshot) { diff --git a/src/tests/integration/test_room.cpp b/src/tests/integration/test_room.cpp index d3e0f29e..6190bf88 100644 --- a/src/tests/integration/test_room.cpp +++ b/src/tests/integration/test_room.cpp @@ -297,6 +297,32 @@ TEST_F(RoomTest, ServerDeletedRoomDisconnectsAndTearsDownLocally) { 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 // destroyed. Because the accessors return std::weak_ptr and the Room is the // sole owner of the participant shared_ptrs, a consumer that caches the handles diff --git a/src/tests/unit/test_room.cpp b/src/tests/unit/test_room.cpp index 327a7b62..e68854cd 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -40,7 +40,7 @@ struct RoomTestAccess { room.connection_state_ = ConnectionState::Connected; room.room_handle_ = std::make_shared(); room.listener_id_ = listener_id; - room.teardown_started_ = false; + room.shutdown_started_ = false; } static bool hasRoomHandle(const Room& room) { @@ -52,6 +52,17 @@ struct RoomTestAccess { 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 @@ -104,7 +115,39 @@ TEST_F(RoomTest, ServerDisconnectTearsDownRoomAndRemovesListener) { 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"; + EXPECT_FALSE(room.disconnect()) << "disconnect after server shutdown must be a no-op"; +} + +TEST_F(RoomTest, ConnectAllowedAfterServerDisconnectShutdown) { + 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); + + ASSERT_EQ(room.connectionState(), ConnectionState::Disconnected); + ASSERT_TRUE(RoomTestAccess::shutdownStarted(room)); + ASSERT_EQ(delegate.count, 1); + + // Room::connect() clears shutdown_started_ before FFI work; mirror that reset and run a + // second synthetic connection cycle to prove the Room is reusable after server shutdown. + RoomTestAccess::clearShutdownForReconnect(room); + EXPECT_FALSE(RoomTestAccess::shutdownStarted(room)); + + listener_calls.store(0, std::memory_order_relaxed); + RoomTestAccess::installConnectedListener(room, listener_calls); + emitFfiEvent(event); + + EXPECT_EQ(listener_calls.load(std::memory_order_relaxed), 1); + EXPECT_EQ(delegate.count, 2) << "second connection cycle should notify onDisconnected again"; + EXPECT_EQ(RoomTestAccess::listenerId(room), 0); + EXPECT_TRUE(RoomTestAccess::shutdownStarted(room)); } TEST_F(RoomTest, ConnectWithoutInitialize) { From a2e6a9e75fcfe7a8435ba35989e548bbf432a2cf Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 13 Jul 2026 22:10:55 -0600 Subject: [PATCH 09/13] Cleanup weird test --- src/tests/unit/test_room.cpp | 86 +++++++++++------------------------- 1 file changed, 27 insertions(+), 59 deletions(-) diff --git a/src/tests/unit/test_room.cpp b/src/tests/unit/test_room.cpp index e68854cd..6224b275 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -91,65 +91,6 @@ class UnitDisconnectTrackingDelegate : public RoomDelegate { } // namespace -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"; -} - -TEST_F(RoomTest, ConnectAllowedAfterServerDisconnectShutdown) { - 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); - - ASSERT_EQ(room.connectionState(), ConnectionState::Disconnected); - ASSERT_TRUE(RoomTestAccess::shutdownStarted(room)); - ASSERT_EQ(delegate.count, 1); - - // Room::connect() clears shutdown_started_ before FFI work; mirror that reset and run a - // second synthetic connection cycle to prove the Room is reusable after server shutdown. - RoomTestAccess::clearShutdownForReconnect(room); - EXPECT_FALSE(RoomTestAccess::shutdownStarted(room)); - - listener_calls.store(0, std::memory_order_relaxed); - RoomTestAccess::installConnectedListener(room, listener_calls); - emitFfiEvent(event); - - EXPECT_EQ(listener_calls.load(std::memory_order_relaxed), 1); - EXPECT_EQ(delegate.count, 2) << "second connection cycle should notify onDisconnected again"; - EXPECT_EQ(RoomTestAccess::listenerId(room), 0); - EXPECT_TRUE(RoomTestAccess::shutdownStarted(room)); -} - TEST_F(RoomTest, ConnectWithoutInitialize) { // Test fixture initializes by default, do this to emulate lack of initialization livekit::shutdown(); @@ -396,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 From 7534b225374831eb407cf1dd3334fd5b8cd532a0 Mon Sep 17 00:00:00 2001 From: Alan George Date: Tue, 14 Jul 2026 19:26:24 -0600 Subject: [PATCH 10/13] Removing shutdown from EOS event --- src/room.cpp | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/room.cpp b/src/room.cpp index 5479712b..e5cebaad 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -1221,8 +1221,6 @@ void Room::onEvent(const FfiEvent& event) { break; } case proto::RoomEvent::kEos: { - (void)shutdown(false, DisconnectReason::Unknown, false); - const RoomEosEvent ev; if (delegate_snapshot) { delegate_snapshot->onRoomEos(*this, ev); From 55299c3eee642cdd63b375904710172d8f9f860f Mon Sep 17 00:00:00 2001 From: Alan George Date: Wed, 15 Jul 2026 10:49:40 -0600 Subject: [PATCH 11/13] Address unrecoverable case --- include/livekit/room.h | 2 -- src/room.cpp | 8 ++--- src/tests/integration/test_room.cpp | 45 +++++++++++++++++++++++++++++ src/tests/unit/test_room.cpp | 12 -------- 4 files changed, 48 insertions(+), 19 deletions(-) diff --git a/include/livekit/room.h b/include/livekit/room.h index 7959ebf3..caf4905b 100644 --- a/include/livekit/room.h +++ b/include/livekit/room.h @@ -356,8 +356,6 @@ class LIVEKIT_API Room { 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); }; diff --git a/src/room.cpp b/src/room.cpp index e5cebaad..560651ef 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -102,7 +102,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"); } - shutdown_started_ = false; connection_state_ = ConnectionState::Reconnecting; } @@ -230,12 +229,11 @@ bool Room::shutdown(bool disconnect_ffi, DisconnectReason reason, bool notify_de const std::scoped_lock g(lock_); 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 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; } - shutdown_started_ = true; handle = std::move(room_handle_); delegate_snapshot = delegate_; local_participant_to_cleanup = std::move(local_participant_); diff --git a/src/tests/integration/test_room.cpp b/src/tests/integration/test_room.cpp index 6190bf88..240e2b83 100644 --- a/src/tests/integration/test_room.cpp +++ b/src/tests/integration/test_room.cpp @@ -26,12 +26,24 @@ #include #include #include +#include #include #include "../common/test_common.h" using namespace std::chrono_literals; +namespace livekit { + +struct RoomTestAccess { + static int listenerId(const Room& room) { + const std::scoped_lock guard(room.lock_); + return room.listener_id_; + } +}; + +} // namespace livekit + namespace livekit::test { // Server-dependent tests - require LIVEKIT_URL and LIVEKIT_TOKEN_A env vars @@ -323,6 +335,39 @@ TEST_F(RoomTest, ReconnectAfterDisconnect) { EXPECT_EQ(delegate.count.load(), 2); } +// Case: a late connection remains recoverable after disconnect raced with connect. +TEST_F(RoomTest, LateConnectionAfterDisconnectRemainsRecoverable) { + ASSERT_TRUE(server_available_) << "LIVEKIT_URL and LIVEKIT_TOKEN_A not set"; + + Room room; + RoomOptions options; + auto connect_future = + std::async(std::launch::async, [&room, this, &options]() { return room.connect(server_url_, token_, options); }); + + // Wait until connect() installs its listener immediately before starting the + // blocking FFI connection, then invalidate that in-flight attempt. + const auto deadline = std::chrono::steady_clock::now() + 5s; + while (RoomTestAccess::listenerId(room) == 0 && connect_future.wait_for(0ms) != std::future_status::ready && + std::chrono::steady_clock::now() < deadline) { + std::this_thread::yield(); + } + + ASSERT_EQ(room.connectionState(), ConnectionState::Reconnecting) + << "connect completed before the test could exercise the disconnect race"; + ASSERT_NE(RoomTestAccess::listenerId(room), 0); + ASSERT_TRUE(room.disconnect()) << "disconnect should claim the in-progress connection"; + + // This preserves the existing behavior on main: the in-flight connect can + // still complete, but the Room must not become permanently unshuttable. + ASSERT_TRUE(connect_future.get()); + ASSERT_EQ(room.connectionState(), ConnectionState::Connected); + ASSERT_FALSE(room.localParticipant().expired()); + + EXPECT_TRUE(room.disconnect()) << "the late connection must remain recoverable"; + EXPECT_EQ(room.connectionState(), ConnectionState::Disconnected); + EXPECT_TRUE(room.localParticipant().expired()); +} + // Verifies that participant handles handed out by Room expire once the Room is // destroyed. Because the accessors return std::weak_ptr and the Room is the // sole owner of the participant shared_ptrs, a consumer that caches the handles diff --git a/src/tests/unit/test_room.cpp b/src/tests/unit/test_room.cpp index 6224b275..fbd97f39 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -40,7 +40,6 @@ struct RoomTestAccess { 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) { @@ -52,17 +51,6 @@ struct RoomTestAccess { 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 From 8416551a0969264eb3e6702ae5fe4d8f01b62299 Mon Sep 17 00:00:00 2001 From: Alan George Date: Wed, 15 Jul 2026 13:38:16 -0600 Subject: [PATCH 12/13] Now test against rust fix for EOS --- client-sdk-rust | 2 +- src/room.cpp | 17 +++++++-- src/tests/integration/test_room.cpp | 13 +++++++ src/tests/unit/test_room.cpp | 56 ++++++++++++++++++++--------- 4 files changed, 68 insertions(+), 20 deletions(-) diff --git a/client-sdk-rust b/client-sdk-rust index dad794d4..afbdabf3 160000 --- a/client-sdk-rust +++ b/client-sdk-rust @@ -1 +1 @@ -Subproject commit dad794d414fda9e8c1de83af1c0f190506a15f8f +Subproject commit afbdabf377b211ae1cf91cc1c39ca19730085a6f diff --git a/src/room.cpp b/src/room.cpp index 560651ef..afab104d 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -1193,8 +1193,19 @@ void Room::onEvent(const FfiEvent& event) { break; } case proto::RoomEvent::kDisconnected: { - const auto reason = toDisconnectReason(re.disconnected().reason()); - (void)shutdown(false, reason, true); + bool should_notify = false; + { + const std::scoped_lock guard(lock_); + // 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; + } + if (should_notify && delegate_snapshot) { + DisconnectedEvent ev; + ev.reason = toDisconnectReason(re.disconnected().reason()); + delegate_snapshot->onDisconnected(*this, ev); + } break; } case proto::RoomEvent::kReconnecting: { @@ -1219,6 +1230,8 @@ void Room::onEvent(const FfiEvent& event) { break; } case proto::RoomEvent::kEos: { + (void)shutdown(false, DisconnectReason::Unknown, false); + const RoomEosEvent ev; if (delegate_snapshot) { delegate_snapshot->onRoomEos(*this, ev); diff --git a/src/tests/integration/test_room.cpp b/src/tests/integration/test_room.cpp index 240e2b83..97e11bc6 100644 --- a/src/tests/integration/test_room.cpp +++ b/src/tests/integration/test_room.cpp @@ -160,17 +160,28 @@ class DisconnectTrackingDelegate : public RoomDelegate { cv_.notify_all(); } + void onRoomEos(Room&, const RoomEosEvent&) override { + eos_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; }); } + bool waitForEos(std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, [this]() { return eos_count.load() > 0; }); + } + DisconnectReason lastReason() const { const std::scoped_lock lock(mutex_); return last_reason_; } std::atomic count{0}; + std::atomic eos_count{0}; private: mutable std::mutex mutex_; @@ -300,9 +311,11 @@ TEST_F(RoomTest, ServerDeletedRoomDisconnectsAndTearsDownLocally) { 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"; + ASSERT_TRUE(delegate.waitForEos(10s)) << "server room deletion should end the room event stream"; 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.eos_count.load(), 1) << "onRoomEos should fire exactly once"; EXPECT_EQ(delegate.lastReason(), DisconnectReason::RoomDeleted); EXPECT_FALSE(room.disconnect()) << "disconnect after server teardown should be a no-op"; diff --git a/src/tests/unit/test_room.cpp b/src/tests/unit/test_room.cpp index fbd97f39..9f658661 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include "../common/ffi_utils.h" #include "ffi.pb.h" @@ -66,14 +67,21 @@ class RoomTest : public ::testing::Test { namespace { -class UnitDisconnectTrackingDelegate : public RoomDelegate { +enum class LifecycleCallback { + Disconnected, + RoomEos, +}; + +class UnitLifecycleTrackingDelegate : public RoomDelegate { public: void onDisconnected(Room&, const DisconnectedEvent& event) override { - ++count; reason = event.reason; + callbacks.push_back(LifecycleCallback::Disconnected); } - int count = 0; + void onRoomEos(Room&, const RoomEosEvent&) override { callbacks.push_back(LifecycleCallback::RoomEos); } + + std::vector callbacks; DisconnectReason reason = DisconnectReason::Unknown; }; @@ -325,31 +333,45 @@ TEST_F(RoomTest, RemoteParticipantLookupBeforeConnect) { << "Looking up participant before connect should return an empty handle"; } -TEST_F(RoomTest, ServerDisconnectTearsDownRoomAndRemovesListener) { +TEST_F(RoomTest, ServerDisconnectNotifiesBeforeEosTearsDownRoom) { Room room; - UnitDisconnectTrackingDelegate delegate; + UnitLifecycleTrackingDelegate 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); + proto::FfiEvent disconnected_event; + auto* disconnected_room_event = disconnected_event.mutable_room_event(); + disconnected_room_event->set_room_handle(0); + disconnected_room_event->mutable_disconnected()->set_reason(proto::ROOM_DELETED); - emitFfiEvent(event); + emitFfiEvent(disconnected_event); EXPECT_EQ(listener_calls.load(std::memory_order_relaxed), 1); + EXPECT_EQ(room.connectionState(), ConnectionState::Connected); + EXPECT_TRUE(RoomTestAccess::hasRoomHandle(room)); + EXPECT_NE(RoomTestAccess::listenerId(room), 0); + ASSERT_EQ(delegate.callbacks.size(), 1); + EXPECT_EQ(delegate.callbacks[0], LifecycleCallback::Disconnected); + EXPECT_EQ(delegate.reason, DisconnectReason::RoomDeleted); + + proto::FfiEvent eos_event; + eos_event.mutable_room_event()->set_room_handle(0); + eos_event.mutable_room_event()->mutable_eos(); + emitFfiEvent(eos_event); + + EXPECT_EQ(listener_calls.load(std::memory_order_relaxed), 2); 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"; + ASSERT_EQ(delegate.callbacks.size(), 2); + EXPECT_EQ(delegate.callbacks[0], LifecycleCallback::Disconnected); + EXPECT_EQ(delegate.callbacks[1], LifecycleCallback::RoomEos); + + emitFfiEvent(eos_event); + EXPECT_EQ(listener_calls.load(std::memory_order_relaxed), 2) << "EOS teardown must unregister the Room listener"; + EXPECT_EQ(delegate.callbacks.size(), 2) << "lifecycle delegates must each fire exactly once"; + EXPECT_FALSE(room.disconnect()) << "disconnect after EOS teardown must be a no-op"; } } // namespace livekit::test From a6a8d20764048e0defe141fbe68c8ca65967b56b Mon Sep 17 00:00:00 2001 From: Alan George Date: Wed, 15 Jul 2026 13:57:42 -0600 Subject: [PATCH 13/13] Cleaner log lines --- client-sdk-rust | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client-sdk-rust b/client-sdk-rust index afbdabf3..bb225e0d 160000 --- a/client-sdk-rust +++ b/client-sdk-rust @@ -1 +1 @@ -Subproject commit afbdabf377b211ae1cf91cc1c39ca19730085a6f +Subproject commit bb225e0dc512df88c2787c335132a9ac18452117