diff --git a/.changeset/engine_task_weak_self_reference.md b/.changeset/engine_task_weak_self_reference.md new file mode 100644 index 000000000..779659db3 --- /dev/null +++ b/.changeset/engine_task_weak_self_reference.md @@ -0,0 +1,15 @@ +--- +livekit: patch +livekit-capture: patch +livekit-ffi: patch +--- + +Fix the RTC engine leaking when a room is dropped without being closed. + +The engine's event task held a strong reference to the engine, while the signal that stops +that task lives inside the engine itself, so the two kept each other alive. An engine +dropped without an explicit `close()` released nothing: the session, both peer connections, +the WebRTC runtime, and the signal client with its open websocket all stayed resident for +the lifetime of the process, and the server kept its half of the session because the socket +was never closed. The task now holds a weak reference and stops on its own once the engine +is gone. diff --git a/livekit/src/rtc_engine/mod.rs b/livekit/src/rtc_engine/mod.rs index 4c4e3deb1..9c9169993 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -17,7 +17,13 @@ use livekit_datatrack::backend as dt; use livekit_protocol as proto; use livekit_signaling::{SignalError, SignalOptions}; use parking_lot::{RwLock, RwLockReadGuard}; -use std::{borrow::Cow, collections::HashSet, fmt::Debug, sync::Arc, time::Duration}; +use std::{ + borrow::Cow, + collections::HashSet, + fmt::Debug, + sync::{Arc, Weak}, + time::Duration, +}; use thiserror::Error; use tokio::sync::{ mpsc, oneshot, Notify, RwLock as AsyncRwLock, RwLockReadGuard as AsyncRwLockReadGuard, @@ -328,6 +334,15 @@ impl RtcEngine { self.inner.close(reason).await } + /// Test-only: returns a probe that reports whether the engine internals have been + /// dropped. `Drop` running is not observable from the outside, and a reference cycle + /// would silently prevent it, so lifecycle tests assert on this instead. + #[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 async fn publish_data( &self, data: proto::DataPacket, @@ -517,8 +532,11 @@ impl EngineInner { // Start initial tasks let (close_tx, close_rx) = oneshot::channel(); - let session_task = - tokio::spawn(Self::engine_task(inner.clone(), session_events, close_rx)); + let session_task = tokio::spawn(Self::engine_task( + Arc::downgrade(&inner), + session_events, + close_rx, + )); inner.running_handle.write().engine_task = Some((session_task, close_tx)); Ok((inner, join_response, engine_rx)) @@ -556,16 +574,26 @@ impl EngineInner { Err(last_error.unwrap()) } + /// Forwards session events to the engine for as long as the engine is alive. + /// + /// Takes a [`Weak`] reference because the engine owns this task: its `JoinHandle` and + /// stop signal live in [`EngineHandle`], inside the very [`EngineInner`] this task would + /// otherwise hold. A strong reference makes the two keep each other alive, so nothing + /// but an explicit `close()` can release them, and merely dropping the engine leaks the + /// session, both peer connections, and the signal client with its open websocket. + /// Upgrading per event also lets the task stop on its own once the engine is gone. async fn engine_task( - self: Arc, + this: Weak, mut session_events: SessionEvents, mut close_rx: oneshot::Receiver<()>, ) { loop { tokio::select! { Some(event) = session_events.recv() => { + let Some(inner) = this.upgrade() else { + break; + }; let debug = format!("{:?}", event); - let inner = self.clone(); let (tx, rx) = oneshot::channel(); let task = tokio::spawn(async move { if let Err(err) = inner.on_session_event(event).await { @@ -1106,7 +1134,7 @@ impl EngineInner { handle.full_reconnect = false; let (close_tx, close_rx) = oneshot::channel(); - let task = tokio::spawn(self.clone().engine_task(session_events, close_rx)); + let task = tokio::spawn(Self::engine_task(Arc::downgrade(self), session_events, close_rx)); handle.engine_task = Some((task, close_tx)); Ok(()) diff --git a/livekit/tests/common/e2e/mod.rs b/livekit/tests/common/e2e/mod.rs index 258c71953..190bb438a 100644 --- a/livekit/tests/common/e2e/mod.rs +++ b/livekit/tests/common/e2e/mod.rs @@ -16,7 +16,7 @@ use anyhow::{Context, Result}; use chrono::Utc; use futures_util::future::try_join_all; use libwebrtc::native::create_random_uuid; -use livekit::{Room, RoomEvent, RoomOptions}; +use livekit::{rtc_engine::RtcEngine, Room, RoomEvent, RoomOptions}; use livekit_token::{AccessToken, VideoGrants}; use std::{env, time::Duration}; use tokio::{ @@ -127,3 +127,27 @@ pub async fn test_rooms_with_options( Ok(rooms) } + +/// Creates a connected [`RtcEngine`] for testing, bypassing the [`Room`] layer. +/// +/// Lifecycle assertions about the engine need it to be the sole owner of its +/// internals. Going through [`Room`] would keep the engine alive through the +/// room's own ownership edges, masking what is being asserted. +pub async fn test_engine() -> Result { + let test_env = TestEnvironment::from_env_or_defaults(); + let room_name = format!("test_room_{}", create_random_uuid()); + + let token = AccessToken::with_api_key(&test_env.api_key, &test_env.api_secret) + .with_ttl(Duration::from_secs(30 * 60)) // 30 minutes + .with_grants(VideoGrants { room_join: true, room: room_name, ..Default::default() }) + .with_identity("p0") + .with_name("Participant 0") + .to_jwt() + .context("Failed to generate JWT")?; + + let (engine, _, _) = RtcEngine::connect(&test_env.server_url, &token, Default::default(), None) + .await + .context("Failed to connect to engine")?; + + Ok(engine) +} diff --git a/livekit/tests/engine_lifecycle_test.rs b/livekit/tests/engine_lifecycle_test.rs new file mode 100644 index 000000000..98602f067 --- /dev/null +++ b/livekit/tests/engine_lifecycle_test.rs @@ -0,0 +1,51 @@ +// 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. + +#[cfg(feature = "__lk-e2e-test")] +use { + anyhow::{Ok, Result}, + common::test_engine, + std::time::Duration, + tokio::time::{self, timeout}, +}; + +mod common; + +/// Dropping the engine without an explicit `close()` must still release its internals. +/// +/// The engine task is owned by the engine (its `JoinHandle` is stored in `EngineHandle`), +/// so it must not hold a strong reference back to what owns it. If it does, the task and +/// the engine keep each other alive and nothing short of `close()` can break the cycle, +/// leaking the session, both peer connections, the signal client and its websocket. +#[cfg(feature = "__lk-e2e-test")] +#[test_log::test(tokio::test)] +async fn test_drop_without_close_releases_engine() -> Result<()> { + let engine = test_engine().await?; + let engine_dropped = engine.drop_probe(); + + drop(engine); + + // Teardown unwinds across several tasks, so poll for release rather than + // asserting immediately or sleeping for an arbitrary fixed duration. + let released = timeout(Duration::from_secs(10), async { + while !engine_dropped() { + time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .is_ok(); + + assert!(released, "engine internals retained after drop"); + Ok(()) +}