From ee2cecf8f047e74602034b5f602f0d53e18b1d40 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Fri, 18 Sep 2026 13:21:05 -0300 Subject: [PATCH] fix(ffi): hold the room weakly from its own RPC method handlers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `register_rpc_method` cloned `Arc` into the handler it registers on the SDK. That handler is stored on the room's own RPC server — `RoomInner` owns the `livekit::Room`, which owns the `RoomSession`, which owns the handler map — so the capture closed a cycle back onto the object that transitively stores it. Nothing unregisters the method during teardown. `RoomSession::close` fails pending client RPCs but never touches `rpc_server.handlers`, and `FfiRoom::close` does not either, so the cycle survives both. `FfiServer::dispose` only clears the handle map, which drops one `Arc` out of a self-sustaining cycle and releases nothing: the room, the engine, its peer connections and the WebRTC runtime stay resident for the rest of the process. That defeats what `dispose()` is documented to do, and repeated connect/dispose cycles leak a full room each time. Only an explicit `UnregisterRpcMethod` per method broke it. Capture a `Weak` and upgrade per invocation, as AGENTS.md requires for a callback stored by the object it captures. An invocation that arrives once the room is gone now fails cleanly instead of resurrecting it. A weak capture alone only breaks the idle cycle, though. Each accepted invocation upgrades it to a strong `Arc`, stores its responder in the room and parks on the matching receiver with no timeout. Disposal removes the client's handles, so the response can never arrive and the handler would stay pending forever, keeping the room alive exactly as the idle cycle did. `FfiRoom::close` now drains `rpc_method_invocation_waiters` and fails each one, after the room is closed so the SDK is no longer delivering invocations that could repopulate the map. Adds a `__lk-e2e-test` feature mirroring the one in `livekit`, and two tests under it. The first registers a method, disposes, and asserts the room is released. The second has a second participant invoke that method, waits for the invocation to reach the handler, never answers it, then disposes and asserts the same. Each fails without its respective change. `serial_test` is added because `FFI_SERVER` is a process-wide singleton that both tests configure and dispose. The drain also marks the room closed under the same lock, and `store_rpc_method_invocation_waiter` refuses entries once that is set. Incoming RPCs run on detached tasks that `RoomSession::close` does not join, so one can reach the handler and upgrade the room after the drain has run; a waiter stored then could never be answered, because the client's handles are gone by that point, and the handler would hold the room open indefinitely. Such an invocation now fails immediately instead. --- .changeset/ffi_rpc_handler_weak_room.md | 13 + Cargo.lock | 1 + livekit-ffi/Cargo.toml | 4 + livekit-ffi/src/server/mod.rs | 3 + livekit-ffi/src/server/participant.rs | 29 +- livekit-ffi/src/server/room.rs | 59 +++- livekit-ffi/src/server/rpc_lifecycle_tests.rs | 265 ++++++++++++++++++ 7 files changed, 367 insertions(+), 7 deletions(-) create mode 100644 .changeset/ffi_rpc_handler_weak_room.md create mode 100644 livekit-ffi/src/server/rpc_lifecycle_tests.rs diff --git a/.changeset/ffi_rpc_handler_weak_room.md b/.changeset/ffi_rpc_handler_weak_room.md new file mode 100644 index 000000000..5feea3349 --- /dev/null +++ b/.changeset/ffi_rpc_handler_weak_room.md @@ -0,0 +1,13 @@ +--- +livekit-ffi: patch +--- + +Fix a room leaking after `dispose()` once an RPC method has been registered. + +The handler registered for an RPC method captured the room strongly, and it is stored on +that same room's RPC server, so the two formed a cycle that nothing unregistered during +teardown. A single registered method made the room outlive `dispose()`, keeping the engine, +its peer connections and the WebRTC runtime resident for the rest of the process. The +handler now captures the room weakly. Closing a room also fails any RPC invocation still +awaiting a response, which would otherwise hold the room open indefinitely because the +client can no longer answer once its handles are gone. diff --git a/Cargo.lock b/Cargo.lock index 28db47b34..f1f9700a6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3744,6 +3744,7 @@ dependencies = [ "parking_lot", "prost 0.14.4", "prost-build 0.14.4", + "serial_test", "soxr-sys", "thiserror 2.0.19", "tokio", diff --git a/livekit-ffi/Cargo.toml b/livekit-ffi/Cargo.toml index 044dd32f2..faa0a6cdc 100644 --- a/livekit-ffi/Cargo.toml +++ b/livekit-ffi/Cargo.toml @@ -15,6 +15,7 @@ native-tls-vendored = ["livekit/native-tls-vendored"] rustls-tls-native-roots = ["livekit/rustls-tls-native-roots"] rustls-tls-webpki-roots = ["livekit/rustls-tls-webpki-roots"] __rustls-tls = ["livekit/__rustls-tls"] +__lk-e2e-test = [] # end-to-end testing with a LiveKit server # Enable tokio-console to debug tasks tracing = ["tokio/tracing", "console-subscriber"] @@ -59,6 +60,9 @@ webrtc-sys-build = { workspace = true } [dev-dependencies] livekit-token = { workspace = true } +# The FFI server is a process-wide singleton, so tests that set it up and dispose it +# must not run concurrently with each other. +serial_test = "3.0" [lib] crate-type = ["lib", "staticlib", "cdylib"] diff --git a/livekit-ffi/src/server/mod.rs b/livekit-ffi/src/server/mod.rs index 43a2dcd30..ecfc32b56 100644 --- a/livekit-ffi/src/server/mod.rs +++ b/livekit-ffi/src/server/mod.rs @@ -57,6 +57,9 @@ mod tests; #[cfg(test)] mod audio_filter_tests; +#[cfg(all(test, feature = "__lk-e2e-test"))] +mod rpc_lifecycle_tests; + #[derive(Clone)] pub struct FfiConfig { pub callback_fn: Arc, diff --git a/livekit-ffi/src/server/participant.rs b/livekit-ffi/src/server/participant.rs index 0ed29b766..59b2a5935 100644 --- a/livekit-ffi/src/server/participant.rs +++ b/livekit-ffi/src/server/participant.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::Arc; +use std::sync::{Arc, Weak}; use livekit::prelude::*; use std::time::Duration; @@ -100,12 +100,27 @@ impl FfiParticipant { }; let local_participant_handle = self.handle.clone(); - let room: Arc = self.room.clone(); + // Weak, not strong: the handler is stored on this room's own RPC server + // (`RoomInner` -> `Room` -> `RoomSession` -> handler map), so capturing the room + // strongly closes a cycle onto the object that stores the handler. Nothing + // unregisters the method during teardown, so that cycle survives both + // `FfiRoom::close` and `FfiServer::dispose`, keeping the room — and with it the + // engine, its peer connections and the WebRTC runtime — alive for the rest of the + // process. Upgrading per invocation also makes a call that arrives while the room + // is going away fail cleanly instead of resurrecting it. + let room: Weak = Arc::downgrade(&self.room); local.register_rpc_method(method.clone(), move |data| { Box::pin({ let room = room.clone(); let method = method.clone(); async move { + let Some(room) = room.upgrade() else { + return Err(RpcError { + code: RpcErrorCode::ApplicationError as u32, + message: "The room has been closed".to_string(), + data: None, + }); + }; forward_rpc_method_invocation( server, room, @@ -318,7 +333,15 @@ async fn forward_rpc_method_invocation( let (tx, rx) = oneshot::channel(); let invocation_id = server.next_id(); - room.store_rpc_method_invocation_waiter(invocation_id, tx); + if !room.store_rpc_method_invocation_waiter(invocation_id, tx) { + // The room is being torn down, so the client can no longer answer. Fail now rather + // than park on a waiter nothing will complete, which would keep the room alive. + return Err(RpcError { + code: RpcErrorCode::ApplicationError as u32, + message: "The room has been closed".to_string(), + data: None, + }); + } let _ = server.send_event( proto::RpcMethodInvocationEvent { diff --git a/livekit-ffi/src/server/room.rs b/livekit-ffi/src/server/room.rs index cc92de40e..3e288c462 100644 --- a/livekit-ffi/src/server/room.rs +++ b/livekit-ffi/src/server/room.rs @@ -85,12 +85,23 @@ pub struct RoomInner { local_publication_lookup: Arc>>, // Used to forward RPC method invocation to the FfiClient and collect their results - rpc_method_invocation_waiters: Mutex>>>, + rpc_method_invocation_waiters: Mutex, // ws url associated with this room url: String, } +/// Pending RPC invocations waiting on a response from the FFI client. +/// +/// `closed` is held under the same lock as `pending` so teardown can drain the map and refuse +/// later arrivals in one step. Without it a handler that upgraded the room just before the +/// drain could insert a waiter afterwards and park on it forever, holding the room open. +#[derive(Default)] +struct RpcMethodInvocationWaiters { + closed: bool, + pending: HashMap>>, +} + const ROOM_EVENT_READY_TIMEOUT: Duration = Duration::from_secs(15); struct Handle { @@ -128,6 +139,15 @@ struct FfiSipDtmfPacket { } impl FfiRoom { + /// Test-only: returns a probe that reports whether the room's internals have been + /// dropped. Callbacks the room installs on the SDK can hold it back, and such a cycle + /// silently prevents `Drop` from ever running, so lifecycle tests assert on this. + #[cfg(feature = "__lk-e2e-test")] + pub fn drop_probe(&self) -> impl Fn() -> bool + Send + Sync + 'static { + let inner = Arc::downgrade(&self.inner); + move || inner.upgrade().is_none() + } + pub fn connect( server: &'static FfiServer, connect: proto::ConnectRequest, @@ -371,6 +391,26 @@ impl FfiRoom { let _ = self.inner.room.close_with_reason(reason.into()).await; + // Fail any RPC invocation still waiting on a response from the FFI client. Each + // waiting handler holds an `Arc` it upgraded from the weak capture in + // `register_rpc_method`, and it parks on the matching receiver with no timeout. The + // client can no longer answer once its handles are gone, so without this the handler + // stays pending forever and keeps the room — and the WebRTC graph behind it — alive + // past `dispose()`. Drained after the room is closed, so the SDK is no longer + // delivering invocations that could repopulate the map. + let waiters = { + let mut waiters = self.inner.rpc_method_invocation_waiters.lock(); + waiters.closed = true; + std::mem::take(&mut waiters.pending) + }; + for (_, waiter) in waiters { + let _ = waiter.send(Err(RpcError { + code: RpcErrorCode::ApplicationError as u32, + message: "The room has been closed".to_string(), + data: None, + })); + } + let handle = self.handle.lock().await.take(); if let Some(handle) = handle { let _ = handle.close_tx.send(()); @@ -864,19 +904,30 @@ impl RoomInner { proto::SendStreamTrailerResponse { async_id } } + /// Registers a waiter for an RPC invocation, unless the room is already closing. + /// + /// Returns `false` once teardown has drained the waiters. Incoming RPCs run on detached + /// tasks that `RoomSession::close` does not join, so one can reach the handler and upgrade + /// the room after the drain; a waiter stored then would never be answered, because the FFI + /// client's handles are gone, and the handler would hold the room open indefinitely. pub fn store_rpc_method_invocation_waiter( &self, invocation_id: u64, waiter: oneshot::Sender>, - ) { - self.rpc_method_invocation_waiters.lock().insert(invocation_id, waiter); + ) -> bool { + let mut waiters = self.rpc_method_invocation_waiters.lock(); + if waiters.closed { + return false; + } + waiters.pending.insert(invocation_id, waiter); + true } pub fn take_rpc_method_invocation_waiter( &self, invocation_id: u64, ) -> Option>> { - return self.rpc_method_invocation_waiters.lock().remove(&invocation_id); + return self.rpc_method_invocation_waiters.lock().pending.remove(&invocation_id); } pub fn set_data_channel_buffered_amount_low_threshold( diff --git a/livekit-ffi/src/server/rpc_lifecycle_tests.rs b/livekit-ffi/src/server/rpc_lifecycle_tests.rs new file mode 100644 index 000000000..190907e2a --- /dev/null +++ b/livekit-ffi/src/server/rpc_lifecycle_tests.rs @@ -0,0 +1,265 @@ +// Copyright 2025 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. + +//! Lifecycle tests for RPC methods registered through the FFI layer. +//! +//! These connect to a real LiveKit server, so they are gated behind `__lk-e2e-test` +//! exactly like the equivalent suites in the `livekit` crate, and expect a +//! `livekit-server --dev` on the usual development endpoint. +//! +//! `FFI_SERVER` is a process-wide singleton that each test configures and disposes, so +//! every test here is `#[serial]`. + +use std::{ + env, + sync::Arc, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use livekit_token::{AccessToken, VideoGrants}; +use serial_test::serial; +use tokio::sync::mpsc::{self, UnboundedReceiver}; + +use super::{participant::FfiParticipant, room::FfiRoom, FfiConfig}; +use crate::{proto, FfiHandleId, FFI_SERVER}; + +struct TestEnvironment { + api_key: String, + api_secret: String, + server_url: String, +} + +impl TestEnvironment { + fn from_env_or_defaults() -> Self { + Self { + api_key: env::var("LIVEKIT_API_KEY").unwrap_or_else(|_| "devkey".into()), + api_secret: env::var("LIVEKIT_API_SECRET").unwrap_or_else(|_| "secret".into()), + server_url: env::var("LIVEKIT_URL").unwrap_or_else(|_| "ws://localhost:7880".into()), + } + } + + fn token(&self, room: &str, identity: &str) -> String { + AccessToken::with_api_key(&self.api_key, &self.api_secret) + .with_ttl(Duration::from_secs(10 * 60)) + .with_grants(VideoGrants { + room_join: true, + room: room.to_owned(), + ..Default::default() + }) + .with_identity(identity) + .with_name(identity) + .to_jwt() + .expect("failed to generate a join token") + } +} + +fn unique_room_name() -> String { + let nanos = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos(); + format!("ffi_rpc_lifecycle_{nanos}") +} + +/// Configures the FFI server and returns the stream of events it emits. +fn setup_server() -> UnboundedReceiver { + let (event_tx, event_rx) = mpsc::unbounded_channel(); + FFI_SERVER.setup(FfiConfig { + callback_fn: Arc::new(move |event| { + let _ = event_tx.send(event); + }), + capture_logs: false, + sdk: "livekit-ffi-test".to_owned(), + sdk_version: "test".to_owned(), + }); + event_rx +} + +/// Connects a room through the FFI layer and returns its room and local participant +/// handles, with the room-event ready handshake already completed. +/// +/// The handshake matters for lifecycle assertions: the connect task parks on it and +/// holds the room while it waits, so leaving it pending would keep the room alive for +/// reasons unrelated to what is being measured. +async fn connect_room( + events: &mut UnboundedReceiver, + url: &str, + token: &str, +) -> (FfiHandleId, FfiHandleId) { + FfiRoom::connect( + &FFI_SERVER, + proto::ConnectRequest { + url: url.to_owned(), + token: token.to_owned(), + options: proto::RoomOptions::default(), + request_async_id: None, + }, + ); + + let handles = tokio::time::timeout(Duration::from_secs(30), async { + loop { + let event = events.recv().await.expect("the ffi event stream closed"); + let Some(proto::ffi_event::Message::Connect(connect)) = event.message else { continue }; + match connect.message { + Some(proto::connect_callback::Message::Result(result)) => { + break (result.room.handle.id, result.local_participant.handle.id) + } + Some(proto::connect_callback::Message::Error(err)) => { + panic!("failed to connect to the room: {err}") + } + None => panic!("the connect callback carried no result"), + } + } + }) + .await + .expect("timed out waiting for the connect callback"); + + FFI_SERVER + .retrieve_handle::(handles.0) + .expect("no room handle") + .ready_for_room_event(); + + handles +} + +/// Registering an RPC method must not make the room impossible to release. +/// +/// The handler is stored on the room's own RPC server: `RoomInner` owns the +/// `livekit::Room`, which owns the `RoomSession`, which owns the handler map. A handler +/// that captures `Arc` therefore closes a cycle back onto the object that +/// transitively stores it, and nothing unregisters the method during teardown — neither +/// `FfiRoom::close` nor `FfiServer::dispose` touches the SDK's handler map. The room then +/// outlives `dispose()`, keeping the engine, its peer connections and the WebRTC runtime +/// resident for the rest of the process, which is precisely what `dispose()` exists to +/// prevent. +#[test] +#[serial] +fn registering_an_rpc_method_does_not_retain_the_room() { + let test_env = TestEnvironment::from_env_or_defaults(); + let room_name = unique_room_name(); + let token = test_env.token(&room_name, "p0"); + let mut events = setup_server(); + + let room_dropped = FFI_SERVER.async_runtime.block_on(async { + let (room_handle, participant_handle) = + connect_room(&mut events, &test_env.server_url, &token).await; + + let room_dropped = FFI_SERVER + .retrieve_handle::(room_handle) + .expect("no room handle") + .drop_probe(); + + FFI_SERVER + .retrieve_handle::(participant_handle) + .expect("no local participant handle") + .register_rpc_method( + &FFI_SERVER, + proto::RegisterRpcMethodRequest { + local_participant_handle: participant_handle, + method: "lifecycle-probe".to_owned(), + }, + ) + .expect("failed to register the rpc method"); + + FFI_SERVER.dispose().await; + room_dropped + }); + + assert!(FFI_SERVER.ffi_handles.is_empty(), "dispose left handles behind"); + assert!(room_dropped(), "the registered RPC handler retained the room after dispose"); +} + +/// An RPC invocation still awaiting its FFI response must not survive disposal. +/// +/// Making the handler capture the room weakly only breaks the *idle* cycle. Each accepted +/// invocation upgrades that weak reference to a strong `Arc`, stores its +/// responder in the room and parks on the matching receiver with no timeout. Disposal +/// removes the client's handles, so the response can never arrive: unless teardown drains +/// the waiters, the handler stays pending forever and keeps the room alive exactly as the +/// idle cycle used to. +#[test] +#[serial] +fn an_unanswered_rpc_invocation_does_not_retain_the_room() { + let test_env = TestEnvironment::from_env_or_defaults(); + let room_name = unique_room_name(); + let callee_token = test_env.token(&room_name, "callee"); + let caller_token = test_env.token(&room_name, "caller"); + let mut events = setup_server(); + + let callee_room_dropped = FFI_SERVER.async_runtime.block_on(async { + let (callee_room, callee_participant) = + connect_room(&mut events, &test_env.server_url, &callee_token).await; + let (_caller_room, caller_participant) = + connect_room(&mut events, &test_env.server_url, &caller_token).await; + + let callee_room_dropped = FFI_SERVER + .retrieve_handle::(callee_room) + .expect("no room handle") + .drop_probe(); + + FFI_SERVER + .retrieve_handle::(callee_participant) + .expect("no callee participant handle") + .register_rpc_method( + &FFI_SERVER, + proto::RegisterRpcMethodRequest { + local_participant_handle: callee_participant, + method: "never-answered".to_owned(), + }, + ) + .expect("failed to register the rpc method"); + + FFI_SERVER + .retrieve_handle::(caller_participant) + .expect("no caller participant handle") + .perform_rpc( + &FFI_SERVER, + proto::PerformRpcRequest { + local_participant_handle: caller_participant, + destination_identity: "callee".to_owned(), + method: "never-answered".to_owned(), + payload: "ping".to_owned(), + // Long enough that the caller's own timeout cannot be what releases + // the room, which would make this assertion meaningless. + response_timeout_ms: Some(120_000), + request_async_id: None, + max_round_trip_latency_ms: None, + }, + ) + .expect("failed to perform the rpc"); + + // Wait until the callee's handler has actually parked on its responder: at this + // point it holds a strong reference to the room. + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let event = events.recv().await.expect("the ffi event stream closed"); + if let Some(proto::ffi_event::Message::RpcMethodInvocation(invocation)) = + event.message + { + assert_eq!(invocation.method, "never-answered"); + break; + } + } + }) + .await + .expect("timed out waiting for the rpc invocation to reach the callee"); + + // Deliberately never send an RpcMethodInvocationResponse. + FFI_SERVER.dispose().await; + callee_room_dropped + }); + + assert!(FFI_SERVER.ffi_handles.is_empty(), "dispose left handles behind"); + assert!( + callee_room_dropped(), + "an RPC invocation awaiting a response that will never arrive retained the room" + ); +}