From 7224d28372cbbdbe15f6e4b1e90a142fe6fc34fa Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Fri, 18 Sep 2026 10:07:54 -0300 Subject: [PATCH 1/2] fix(rtc_engine): hold a weak reference to the engine from its own task `engine_task` took an `Arc`, but the task's `JoinHandle` and the oneshot sender that stops it both live in `EngineHandle`, inside that same `EngineInner`. The task therefore kept alive the only thing that could ever stop it. An engine dropped without an explicit `close()` released nothing: the session, both peer connections, the signal client with its open websocket, and the task itself stayed resident for the lifetime of the process, and the SFU kept its half of the session because the socket was never closed. `RtcSession` already avoids this by keeping `close_tx` in `SessionHandle`, outside the `Arc`'d inner. Rather than restructure the handle here, take a `Weak` and upgrade per event, which is what AGENTS.md prescribes for a task whose owner stores it. The task now also stops on its own once the engine is gone, instead of depending solely on the close signal. Adds an e2e test asserting the internals are released after a drop without `close()`, and a `test_engine()` harness helper so engine lifecycle can be observed without the room's own ownership edges masking it. --- livekit/src/rtc_engine/mod.rs | 40 +++++++++++++++++--- livekit/tests/common/e2e/mod.rs | 26 ++++++++++++- livekit/tests/engine_lifecycle_test.rs | 51 ++++++++++++++++++++++++++ 3 files changed, 110 insertions(+), 7 deletions(-) create mode 100644 livekit/tests/engine_lifecycle_test.rs 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(()) +} From da7f4ad382a0692b3216d2635ac163611bb1db2b Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Fri, 18 Sep 2026 11:45:13 -0300 Subject: [PATCH 2/2] Add changeset --- .changeset/engine_task_weak_self_reference.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 .changeset/engine_task_weak_self_reference.md 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.