fix(rtc_engine): hold a weak reference to the engine from its own task

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

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<Self>,
+        this: Weak<Self>,
         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(())

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<RtcEngine> {
+    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)
+}

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(())
+}