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" + ); +}