-
Notifications
You must be signed in to change notification settings - Fork 33
Bugfix: Server-initiated room disconnect shutdown fix #205
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
8f5433c
d5dda8c
0b9ea06
9effbcb
738693b
0672226
7810fa3
9f56644
83c403b
a2e6a9e
e8569e3
7534b22
55299c3
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -212,7 +212,10 @@ bool Room::connect(const std::string& url, const std::string& token, const RoomO | |
|
|
||
| bool Room::disconnect(DisconnectReason reason) { | ||
| TRACE_EVENT0("livekit", "Room::disconnect"); | ||
| return shutdown(true, reason, true); | ||
| } | ||
|
|
||
| bool Room::shutdown(bool disconnect_ffi, DisconnectReason reason, bool notify_delegate) { | ||
| std::shared_ptr<FfiHandle> handle; | ||
| RoomDelegate* delegate_snapshot = nullptr; | ||
| std::shared_ptr<LocalParticipant> local_participant_to_cleanup; | ||
|
|
@@ -224,58 +227,82 @@ bool Room::disconnect(DisconnectReason reason) { | |
|
|
||
| { | ||
| const std::scoped_lock<std::mutex> g(lock_); | ||
| if (connection_state_ == ConnectionState::Disconnected) { | ||
| // Already torn down (or never connected). Nothing to do. | ||
| const bool has_room_state = connection_state_ != ConnectionState::Disconnected || listener_id_ != 0 || | ||
| room_handle_ || local_participant_ || !remote_participants_.empty(); | ||
| // Return false for a no-op so callers can tell whether this call claimed the | ||
| // room state and performed cleanup. Matches disconnect()'s documented contract. | ||
| if (!has_room_state) { | ||
| return false; | ||
| } | ||
| handle = room_handle_; | ||
| handle = std::move(room_handle_); | ||
| delegate_snapshot = delegate_; | ||
| // Take ownership of everything under the lock so the kEos handler (which | ||
| // also tries to move it out) loses any race here — only one teardown | ||
| // path operates on this state. | ||
| local_participant_to_cleanup = std::move(local_participant_); | ||
| remote_participants_to_clear = std::move(remote_participants_); | ||
| e2ee_manager_to_clear = std::move(e2ee_manager_); | ||
| text_stream_readers_to_clear = std::move(text_stream_readers_); | ||
| byte_stream_readers_to_clear = std::move(byte_stream_readers_); | ||
| listener_to_remove = listener_id_; | ||
| listener_id_ = 0; | ||
| room_handle_.reset(); | ||
| // Flip state immediately so the in-flight Disconnected room-event we'll | ||
| // get back doesn't double-fire onDisconnected. Mirrors Python's | ||
| // Room.disconnect() | ||
| connection_state_ = ConnectionState::Disconnected; | ||
| } | ||
|
|
||
| // Drain in-flight RPC handlers BEFORE telling Rust to tear down the room. | ||
| // Mirrors client-sdk-python's Room.disconnect() ordering | ||
| bool shutdown_ok = true; | ||
| if (local_participant_to_cleanup) { | ||
| local_participant_to_cleanup->shutdown(); | ||
| try { | ||
| local_participant_to_cleanup->shutdown(); | ||
| } catch (const std::exception& e) { | ||
| LK_LOG_ERROR("Room shutdown: local participant shutdown failed: {}", e.what()); | ||
| shutdown_ok = false; | ||
| } catch (...) { | ||
| LK_LOG_ERROR("Room shutdown: local participant shutdown failed: unknown exception"); | ||
| shutdown_ok = false; | ||
| } | ||
| } | ||
|
|
||
| // Tell the FFI to close the room and wait for the callback. If this fails | ||
| // we still complete local-side teardown below | ||
| bool ffi_ok = true; | ||
| if (handle) { | ||
| if (disconnect_ffi && 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 +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 shutdown_ok; | ||
| } | ||
|
|
||
| RoomInfoData Room::roomInfo() const { | ||
|
|
@@ -1168,22 +1193,8 @@ void Room::onEvent(const FfiEvent& event) { | |
| break; | ||
| } | ||
| case proto::RoomEvent::kDisconnected: { | ||
| // If disconnect() was driven from our side, it already flipped state | ||
| // to Disconnected and fired the delegate; skip the duplicate here. | ||
| bool already_disconnected = false; | ||
| { | ||
| const std::scoped_lock<std::mutex> guard(lock_); | ||
| already_disconnected = (connection_state_ == ConnectionState::Disconnected); | ||
| connection_state_ = ConnectionState::Disconnected; | ||
| } | ||
| if (already_disconnected) { | ||
| break; | ||
| } | ||
| DisconnectedEvent ev; | ||
| ev.reason = toDisconnectReason(re.disconnected().reason()); | ||
| if (delegate_snapshot) { | ||
| delegate_snapshot->onDisconnected(*this, ev); | ||
| } | ||
| const auto reason = toDisconnectReason(re.disconnected().reason()); | ||
| (void)shutdown(false, reason, true); | ||
| break; | ||
|
Comment on lines
1195
to
1198
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🔴 Server-initiated disconnect silently suppresses the end-of-stream callback to the application The listener is removed during server-disconnect cleanup ( Impact: Applications relying on the end-of-stream callback for final cleanup after a server-initiated disconnect will silently miss it. Mechanism: kDisconnected removes the listener before kEos arrivesIn the old code, In the new code,
Because the listener is now removed, when Rust later sends the The old event sequence for server disconnect was: Was this helpful? React with 👍 or 👎 to provide feedback.
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. hard to image apps would be using kEos rather than kDisconnected, i guess something to consider |
||
| } | ||
| case proto::RoomEvent::kReconnecting: { | ||
|
|
@@ -1208,57 +1219,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<LocalParticipant> old_local_participant; | ||
| std::unordered_map<std::string, std::shared_ptr<RemoteParticipant>> old_remote_participants; | ||
| std::shared_ptr<FfiHandle> old_room_handle; | ||
| std::shared_ptr<E2EEManager> old_e2ee_manager; | ||
| std::unordered_map<std::string, std::shared_ptr<TextStreamReader>> old_text_readers; | ||
| std::unordered_map<std::string, std::shared_ptr<ByteStreamReader>> old_byte_readers; | ||
|
|
||
| { | ||
| const std::scoped_lock<std::mutex> guard(lock_); | ||
| listener_to_remove = listener_id_; | ||
| listener_id_ = 0; | ||
|
|
||
| // Reset connection state | ||
| connection_state_ = ConnectionState::Disconnected; | ||
|
|
||
| // Move state out for cleanup outside lock | ||
| old_local_participant = std::move(local_participant_); | ||
| old_remote_participants = std::move(remote_participants_); | ||
| old_room_handle = std::move(room_handle_); | ||
| old_e2ee_manager = std::move(e2ee_manager_); | ||
| old_text_readers = std::move(text_stream_readers_); | ||
| old_byte_readers = std::move(byte_stream_readers_); | ||
| } | ||
|
|
||
| // Drain in-flight RPC invocations before destroying the local | ||
| // participant's FFI handle. Mirrors the ordering in disconnect(); | ||
| // without this, a listener-thread RPC handler can race with handle | ||
| // disposal and send to a dead handle → INVALID_HANDLE → terminate. | ||
| if (old_local_participant) { | ||
| old_local_participant->shutdown(); | ||
| } | ||
|
|
||
| // Remove listener outside lock | ||
| if (listener_to_remove != 0) { | ||
| FfiClient::instance().removeListener(listener_to_remove); | ||
| } | ||
|
|
||
| if (old_local_participant) { | ||
| old_local_participant->shutdown(); | ||
| } | ||
|
|
||
| // old_* state is destroyed here when going out of scope | ||
|
|
||
| const RoomEosEvent ev; | ||
| if (delegate_snapshot) { | ||
| delegate_snapshot->onRoomEos(*this, ev); | ||
|
Comment on lines
1221
to
1224
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🔴 End-of-stream event no longer cleans up room resources, causing leaks when no prior disconnect event arrives All teardown logic was removed from the end-of-stream handler ( Impact: The room leaks FFI resources and its event listener, and becomes permanently stuck in a stale state that cannot be recovered or reconnected. Mechanism: kEos teardown was removed but not replaced with a shutdown() callThe old code at The new The (Refers to lines 1221-1226) Was this helpful? React with 👍 or 👎 to provide feedback. |
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| /* | ||
| * Copyright 2026 LiveKit | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| #pragma once | ||
|
|
||
| #include <gtest/gtest.h> | ||
|
|
||
| #include <cstdint> | ||
| #include <string> | ||
|
|
||
| #include "ffi.pb.h" | ||
| #include "ffi_client.h" | ||
|
|
||
| namespace livekit::test { | ||
|
|
||
| /// Serializes and dispatches a synthetic FFI event through the real callback entry point. | ||
| /// Defined in this header for use across different tests. | ||
| inline void emitFfiEvent(const proto::FfiEvent& event) { | ||
| std::string bytes; | ||
| ASSERT_TRUE(event.SerializeToString(&bytes)); | ||
| ffiEventCallback(reinterpret_cast<const std::uint8_t*>(bytes.data()), bytes.size()); | ||
| } | ||
|
|
||
| } // namespace livekit::test |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
should we return true if nothing to do or it is being torndown ?
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm open to changing this, right now our public API for
disconnect()states:Which this matches,
false== noop. I get how that could be confusing, but I think the idea istrueis when the disconnect actually happenedThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That is fair. We can keep the current behavior to avoid breaking changes.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
FWIW we have a public
connectionStateso apps can check the state before trying to disconnect