Drain RPC invocations on kEos + add lifecycle tracing

The kEos RoomEvent handler moves out and destroys the local participant
without first calling shutdown(). If the listener thread is mid-RPC-
invocation when kEos arrives, the participant's FfiHandle is dropped
while a handler is still in flight; the handler's later sendRequest
then hits an invalid Rust handle and the uncaught std::runtime_error
on the listener thread aborts the process. Mirror the ordering used
in disconnect(): call shutdown() on the moved-out participant before
letting it destruct.

Add INFO-level tracing at the key lifecycle points (~Room, disconnect,
kDisconnected, kEos, FfiHandle drops, INVALID_HANDLE error path) and
DEBUG tracing on the RPC invocation hot path. Bump RUST_LOG in the
integration-test CI step so the Rust side's "handle not found" error
message lands in the log.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: alan-george-lk

.github/workflows/tests.yml

@@ -261,7 +261,11 @@ jobs:
         timeout-minutes: 5
         shell: bash
         env:
-          RUST_LOG: "metrics=debug"
+          # Bump livekit_ffi to debug so handle-table errors (the exact Rust
+          # message that produces our INVALID_HANDLE → terminate path) land
+          # in the CI log. Keep livekit at info to avoid drowning out the
+          # interesting lines.
+          RUST_LOG: "metrics=debug,livekit_ffi=debug,livekit=info"
         run: |
           set -euo pipefail
           source .token_helpers/set_data_track_test_tokens.bash

src/ffi_client.cpp

@@ -202,6 +202,10 @@ proto::FfiResponse FfiClient::sendRequest(const proto::FfiRequest& request) cons
   const FfiHandleId handle =
       livekit_ffi_request(reinterpret_cast<const uint8_t*>(bytes.data()), bytes.size(), &resp_ptr, &resp_len);
   if (handle == INVALID_HANDLE) {
+    LK_LOG_ERROR(
+        "FfiClient::sendRequest: livekit_ffi_request returned INVALID_HANDLE; "
+        "request.message_case={}, bytes_len={}",
+        static_cast<int>(request.message_case()), bytes.size());
     throw std::runtime_error("failed to send request, received an invalid handle");
   }
 

src/ffi_handle.cpp

@@ -17,6 +17,7 @@
 #include "livekit/ffi_handle.h"
 
 #include "livekit_ffi.h"
+#include "lk_log.h"
 
 namespace livekit {
 
@@ -35,6 +36,7 @@ FfiHandle& FfiHandle::operator=(FfiHandle&& other) noexcept {
 
 void FfiHandle::reset(uintptr_t new_handle) noexcept {
   if (handle_) {
+    LK_LOG_INFO("FfiHandle::reset: dropping handle {} (replacement={})", handle_, new_handle);
     livekit_ffi_drop_handle(handle_);
   }
   handle_ = new_handle;

src/local_participant.cpp

@@ -370,13 +370,17 @@ void LocalParticipant::shutdown() {
 void LocalParticipant::handleRpcMethodInvocation(uint64_t invocation_id, const std::string& method,
                                                  const std::string& request_id, const std::string& caller_identity,
                                                  const std::string& payload, double response_timeout_sec) {
+  LK_LOG_DEBUG("LocalParticipant::handleRpcMethodInvocation: entry (handle={}, invocation_id={}, method={})",
+               ffiHandleId(), invocation_id, method);
   // Capture shared state so it outlives LocalParticipant if needed
   auto state = rpc_state_;
 
   // Track this invocation and check if we're shutting down
   {
     const std::scoped_lock<std::mutex> lock(state->mutex);
     if (state->shutting_down) {
+      LK_LOG_DEBUG("LocalParticipant::handleRpcMethodInvocation: shutting_down, skipping (invocation_id={})",
+                   invocation_id);
       // Already shutting down, don't process new invocations
       return;
     }
@@ -423,14 +427,19 @@ void LocalParticipant::handleRpcMethodInvocation(uint64_t invocation_id, const s
   {
     const std::scoped_lock<std::mutex> lock(state->mutex);
     if (state->shutting_down) {
+      LK_LOG_DEBUG(
+          "LocalParticipant::handleRpcMethodInvocation: shutdown during handler, skipping response "
+          "(invocation_id={})",
+          invocation_id);
       // Shutdown started, don't send response - handle may be invalid
       return;
     }
   }
 
   FfiRequest req;
   auto* msg = req.mutable_rpc_method_invocation_response();
-  msg->set_local_participant_handle(ffiHandleId());
+  const auto handle_id_at_send = ffiHandleId();
+  msg->set_local_participant_handle(handle_id_at_send);
   msg->set_invocation_id(invocation_id);
   if (response_error.has_value()) {
     auto* err_proto = msg->mutable_error();
@@ -439,6 +448,8 @@ void LocalParticipant::handleRpcMethodInvocation(uint64_t invocation_id, const s
   if (response_payload.has_value()) {
     msg->set_payload(*response_payload);
   }
+  LK_LOG_DEBUG("LocalParticipant::handleRpcMethodInvocation: sending response (handle={}, invocation_id={})",
+               handle_id_at_send, invocation_id);
   FfiClient::instance().sendRequest(req);
 }
 

src/room.cpp

@@ -76,6 +76,7 @@ void readyForRoomEvent(std::uint64_t room_handle) {
 Room::Room() : subscription_thread_dispatcher_(std::make_unique<SubscriptionThreadDispatcher>()) {}
 
 Room::~Room() {
+  LK_LOG_INFO("Room::~Room: entry (this={})", static_cast<const void*>(this));
   // Issue a graceful disconnect so the server sees us leave instead of
   // timing out (RAII expectation; see issue #118). disconnect() does the
   // full teardown including subscription threads, listener, and local
@@ -245,6 +246,7 @@ bool Room::Connect(const std::string& url, const std::string& token, const RoomO
 
 bool Room::disconnect(DisconnectReason reason) {
   TRACE_EVENT0("livekit", "Room::disconnect");
+  LK_LOG_INFO("Room::disconnect: entry (this={}, reason={})", static_cast<const void*>(this), static_cast<int>(reason));
 
   // Hold onto this in case the
   auto prev_connection_state = connection_state_;
@@ -254,6 +256,7 @@ bool Room::disconnect(DisconnectReason reason) {
   {
     const std::scoped_lock<std::mutex> g(lock_);
     if (connection_state_ == ConnectionState::Disconnected) {
+      LK_LOG_INFO("Room::disconnect: already disconnected, returning false (this={})", static_cast<const void*>(this));
       return false;
     }
     handle = room_handle_;
@@ -1228,6 +1231,8 @@ void Room::onEvent(const FfiEvent& event) {
           break;
         }
         case proto::RoomEvent::kDisconnected: {
+          LK_LOG_INFO("Room::onFfiEvent: kDisconnected received (this={}, reason={})", static_cast<const void*>(this),
+                      static_cast<int>(re.disconnected().reason()));
           // 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;
@@ -1261,6 +1266,7 @@ void Room::onEvent(const FfiEvent& event) {
           break;
         }
         case proto::RoomEvent::kEos: {
+          LK_LOG_INFO("Room::onFfiEvent: kEos received (this={})", static_cast<const void*>(this));
           if (subscription_thread_dispatcher_) {
             subscription_thread_dispatcher_->stopAll();
           }
@@ -1293,6 +1299,16 @@ void Room::onEvent(const FfiEvent& event) {
             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) {
+            LK_LOG_INFO("Room::onFfiEvent: kEos shutting down local participant (handle={})",
+                        old_local_participant->ffiHandleId());
+            old_local_participant->shutdown();
+          }
+
           // Remove listener outside lock
           if (listener_to_remove != 0) {
             FfiClient::instance().removeListener(listener_to_remove);