Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .token_helpers/set_data_track_test_tokens.bash
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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"
Expand Down
5 changes: 4 additions & 1 deletion include/livekit/room.h
Original file line number Diff line number Diff line change
Expand Up @@ -330,7 +330,7 @@ class LIVEKIT_API Room {
void removeOnDataFrameCallback(DataFrameCallbackId id);

private:
friend class RoomCallbackTest;
friend struct RoomTestAccess;

mutable std::mutex lock_;
ConnectionState connection_state_ = ConnectionState::Disconnected;
Expand All @@ -355,5 +355,8 @@ class LIVEKIT_API Room {
int listener_id_{0};

void onEvent(const proto::FfiEvent& event);

// Shared shutdown path for explicit disconnect, server disconnect, EOS, and destruction.
bool shutdown(bool disconnect_ffi, DisconnectReason reason, bool notify_delegate);
};
} // namespace livekit
148 changes: 54 additions & 94 deletions src/room.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Copy link
Copy Markdown
Collaborator

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 ?

@alan-george-lk alan-george-lk Jul 14, 2026

Copy link
Copy Markdown
Collaborator Author

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:

@returns true if the graceful disconnect succeeds; false if the room was already disconnected (no-op) or the graceful disconnect fails.

Which this matches, false == noop. I get how that could be confusing, but I think the idea is true is when the disconnect actually happened

Copy link
Copy Markdown
Collaborator

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.

Copy link
Copy Markdown
Collaborator

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 connectionState so apps can check the state before trying to disconnect

}
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 {
Expand All @@ -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 {
Expand Down Expand Up @@ -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

@devin-ai-integration devin-ai-integration Bot Jul 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The 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 (shutdown(false, reason, true) at src/room.cpp:1199) before the Rust FFI sends the follow-up end-of-stream event, so the onRoomEos delegate callback is never delivered.

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 arrives

In the old code, kDisconnected (at the old src/room.cpp:1170–1186) only set connection_state_ to Disconnected and fired onDisconnected. The FFI listener remained registered, so the subsequent kEos event from Rust would still be dispatched to Room::onEvent(), which performed full state cleanup and called delegate_snapshot->onRoomEos().

In the new code, kDisconnected calls shutdown(false, reason, true) which:

  1. Moves out all room state (src/room.cpp:239–248)
  2. Removes the FFI listener via FfiClient::instance().removeListener() (src/room.cpp:290)
  3. Fires onDisconnected (src/room.cpp:311)

Because the listener is now removed, when Rust later sends the kEos event, FfiClient::pushEvent finds no matching listener slot for this Room. The kEos case at src/room.cpp:1223–1230 is never reached, and onRoomEos is never called.

The old event sequence for server disconnect was: kDisconnectedonDisconnectedkEos → cleanup + onRoomEos.
The new sequence is: kDisconnected → cleanup + onDisconnected → (kEos silently dropped).

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The 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: {
Expand All @@ -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

Copy link
Copy Markdown

Choose a reason for hiding this comment

The 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 (kEos at src/room.cpp:1221), so if the stream ends without a preceding disconnect event, the room's internal state (FFI handles, participants, listener registration) is never released.

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() call

The old code at kEos performed full teardown: it moved out local_participant_, remote_participants_, room_handle_, e2ee_manager_, stream readers, called local_participant->shutdown(), and removed the FFI listener via removeListener(). The new code at src/room.cpp:1221-1226 only fires the delegate callback and does nothing else.

The new shutdown() method's own comment at include/livekit/room.h:359 says it is the "Shared shutdown path for explicit disconnect, server disconnect, EOS, and destruction," confirming the intent was for EOS to use it. But the kEos case never calls shutdown().

The kDisconnected handler at src/room.cpp:1195-1198 does call shutdown(), but if kEos arrives without a preceding kDisconnected (e.g., abnormal server termination, or a protocol edge case), the room handle, local participant, remote participants, E2EE manager, stream readers, and the FFI listener all leak.

(Refers to lines 1221-1226)

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

Expand Down
37 changes: 37 additions & 0 deletions src/tests/common/ffi_utils.h
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
Loading
Loading