fix(ffi): hold the room weakly from its own RPC method handlers

.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.

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"]

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<dyn Fn(FfiEvent) + Send + Sync>,

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<RoomInner> = 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<RoomInner> = 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 {

livekit-ffi/src/server/room.rs

@@ -85,12 +85,23 @@ pub struct RoomInner {
     local_publication_lookup: Arc<Mutex<HashMap<TrackSid, FfiHandleId>>>,
 
     // Used to forward RPC method invocation to the FfiClient and collect their results
-    rpc_method_invocation_waiters: Mutex<HashMap<u64, oneshot::Sender<Result<String, RpcError>>>>,
+    rpc_method_invocation_waiters: Mutex<RpcMethodInvocationWaiters>,
 
     // 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<u64, oneshot::Sender<Result<String, RpcError>>>,
+}
+
 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<RoomInner>` 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<Result<String, RpcError>>,
-    ) {
-        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<oneshot::Sender<Result<String, RpcError>>> {
-        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(