From 1357589bb59e3a0efd63ed4e41025fcea10797c2 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Fri, 18 Sep 2026 13:19:30 -0300 Subject: [PATCH] fix(room): release remote participants when the room is closed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `RoomSession::close` unpublished the local participant's tracks and then dropped the remote participant map. Dropping the map does not release the participants: every remote publication registers callbacks that hold the participant which owns them — `on_subscribed`/`on_unsubscribed` capture the `RemoteParticipant`, and `on_muted`/`on_unmuted` capture it through the publication map — so each remote participant is kept alive by a cycle through its own publications. A leaked participant holds `Arc`, so it pins the whole engine, its peer connections and the WebRTC runtime: exactly what commit 16530d20 set out to release. Each connect/close cycle leaked another one. `handle_participant_disconnect` already unregisters these when a peer actually leaves, but it also emits ParticipantDisconnected and TrackUnpublished. The peers have not left here — the room is closing — so unregister directly instead, leaving the event stream unchanged. Two ordering constraints make this correct: - The teardown runs after `room_handle` has been joined. `room_task` is what inserts remote participants, so draining the map any earlier would let an engine event queued behind the close repopulate it, and those participants would never be visited. - Each publication is unregistered before its track is detached. `RemoteTrackPublication::set_track(None)` invokes the unsubscribe handler *before* clearing it, and that handler dispatches TrackUnsubscribed carrying strong clones of the participant and track. Clearing the handlers first keeps teardown silent, so a receiver the application has stopped draining cannot pin the participant through a queued event. `test_close_releases_room_session` did not catch any of this because it runs with a single participant and no tracks, so neither cycle is ever formed. The new test uses two participants with a published, subscribed track, keeps its event receiver alive and undrained across the close, and asserts via a drop probe that the remote participant's internals are released. The teardown is idempotent and runs twice: once before `close` can suspend, and again after the task joins. `close` takes the room handle first, so a caller that bounds it with a timeout and drops the future cannot retry, and a teardown placed only after the joins would be skipped for good. Running it up front keeps a cancelled close correct, and running it again afterwards catches anything `room_task` inserted in between. --- .../close_releases_remote_participants.md | 13 ++ livekit/src/room/mod.rs | 34 +++++ .../room/participant/remote_participant.rs | 9 ++ livekit/tests/room_test.rs | 131 +++++++++++++++++- 4 files changed, 183 insertions(+), 4 deletions(-) create mode 100644 .changeset/close_releases_remote_participants.md diff --git a/.changeset/close_releases_remote_participants.md b/.changeset/close_releases_remote_participants.md new file mode 100644 index 000000000..7bbb1f638 --- /dev/null +++ b/.changeset/close_releases_remote_participants.md @@ -0,0 +1,13 @@ +--- +livekit: patch +livekit-capture: patch +livekit-ffi: patch +--- + +Fix remote participants leaking when a room is closed. + +Closing a room dropped its remote participant map without unregistering the callbacks each +remote publication installs. Those callbacks hold the participant that owns them, so every +remote participant was kept alive by a cycle through its own publications — and each one +holds the RTC engine, pinning the peer connections and the WebRTC runtime with it. Closing +a room now tears those participants down, leaving the event stream unchanged. diff --git a/livekit/src/room/mod.rs b/livekit/src/room/mod.rs index 21d1c0060..074427201 100644 --- a/livekit/src/room/mod.rs +++ b/livekit/src/room/mod.rs @@ -1179,9 +1179,40 @@ impl RoomSession { Ok(()) } + /// Unregisters the callbacks that keep remote participants alive, and empties the map. + /// + /// Dropping the map is not enough: each remote publication registers callbacks that hold + /// the participant which owns them, so the participant — and the `Arc` it holds + /// — survives the room unless those callbacks are unregistered. + /// + /// Unregister before detaching the track, not after: `RemoteTrackPublication::set_track(None)` + /// invokes the unsubscribe handler *before* clearing, and that handler dispatches + /// `TrackUnsubscribed` carrying strong clones of the participant and track. Clearing the + /// handlers first keeps teardown silent, so a receiver the application has stopped draining + /// cannot pin the participant through a queued event. `handle_participant_disconnect` is + /// deliberately not reused for the same reason: these peers have not left, the room is closing. + /// + /// Idempotent, so `close` can call it both before it suspends and after its tasks are joined. + fn release_remote_participants(&self) { + let remote_participants = std::mem::take(&mut *self.remote_participants.write()); + for participant in remote_participants.into_values() { + for (sid, publication) in participant.track_publications() { + participant.remove_publication(&sid); + publication.set_track(None); + } + } + } + async fn close(&self, reason: DisconnectReason) -> RoomResult<()> { let Some(handle) = self.handle.lock().await.take() else { Err(RoomError::AlreadyClosed)? }; + // Release the remote participants before anything below can suspend. `close` takes the + // handle, so a caller that bounds it with a timeout and drops the future cannot retry: + // a later call returns `AlreadyClosed`. Doing this first means a cancelled close still + // unlinks the callback cycles. It runs again after the task joins, because `room_task` + // can insert a participant in the meantime. + self.release_remote_participants(); + // remove published tracks for (sid, _) in self.local_participant.track_publications().iter() { let _ = self.local_participant.unpublish_track(sid).await; @@ -1201,6 +1232,9 @@ impl RoomSession { let _ = handle.remote_dt_task.await; let _ = handle.room_handle.await; + // Again, to catch anything `room_task` inserted after the drain above. + self.release_remote_participants(); + self.dispatcher.clear(); Ok(()) } diff --git a/livekit/src/room/participant/remote_participant.rs b/livekit/src/room/participant/remote_participant.rs index d6d896f30..98699ae9d 100644 --- a/livekit/src/room/participant/remote_participant.rs +++ b/livekit/src/room/participant/remote_participant.rs @@ -112,6 +112,15 @@ impl RemoteParticipant { self.inner.track_publications.read().clone() } + /// Test-only: returns a probe that reports whether this participant's internals have + /// been dropped. Each of its publications registers callbacks that hold the + /// participant, so teardown has to unregister them; `Drop` alone never runs. + #[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(crate) async fn add_subscribed_media_track( &self, sid: TrackSid, diff --git a/livekit/tests/room_test.rs b/livekit/tests/room_test.rs index 5e05f9246..4e4b78ec7 100644 --- a/livekit/tests/room_test.rs +++ b/livekit/tests/room_test.rs @@ -14,12 +14,15 @@ #[cfg(feature = "__lk-e2e-test")] use { - anyhow::{Ok, Result}, + anyhow::{anyhow, Ok, Result}, chrono::{TimeDelta, TimeZone, Utc}, - common::test_rooms, + common::{ + test_rooms, + video::{SolidColorParams, SolidColorTrack}, + }, libwebrtc::prelude::PeerConnectionState, - livekit::{ConnectionState, ParticipantKind, RoomEvent}, - std::time::Duration, + livekit::{options::VideoCodec, ConnectionState, ParticipantKind, RoomEvent}, + std::{sync::Arc, time::Duration}, tokio::time::{self, timeout}, }; @@ -140,3 +143,123 @@ async fn test_close_releases_room_session() -> Result<()> { assert!(session_dropped(), "room callbacks retained the room session after close"); Ok(()) } + +/// `close()` must tear down remote participants, not only the local one. +/// +/// Every remote publication registers callbacks that hold the participant which owns +/// them: `on_subscribed`/`on_unsubscribed` capture the `RemoteParticipant`, and +/// `on_muted`/`on_unmuted` capture it through the publication map. Those are reference +/// cycles, so a remote participant — and the `Arc` it holds — outlives the +/// room unless teardown unregisters them. Only the disconnect path does that, and +/// `close()` never walks the remote participants. +#[cfg(feature = "__lk-e2e-test")] +#[test_log::test(tokio::test)] +async fn test_close_releases_remote_participants() -> Result<()> { + let mut rooms = test_rooms(2).await?; + let (sub_room, mut sub_events) = rooms.remove(0); + let (pub_room, _pub_events) = rooms.remove(0); + + // The cycles only exist once the remote participant has a publication. + let mut solid_track = SolidColorTrack::new( + Arc::new(pub_room), + SolidColorParams { width: 320, height: 240, luma: 128 }, + ); + solid_track.publish(VideoCodec::VP8, false).await?; + + timeout(Duration::from_secs(15), async { + loop { + match sub_events.recv().await { + Some(RoomEvent::TrackSubscribed { .. }) => break Ok(()), + Some(_) => continue, + None => break Err(anyhow!("event stream ended before the track was subscribed")), + } + } + }) + .await??; + + let remote = sub_room + .remote_participants() + .into_values() + .next() + .ok_or_else(|| anyhow!("subscriber never saw the publisher"))?; + let remote_dropped = remote.drop_probe(); + drop(remote); + + // `sub_events` is deliberately kept alive and undrained across the close. Teardown + // must not dispatch anything carrying the participant: a queued event holds strong + // clones, so a receiver the application has stopped polling would pin the participant + // just as effectively as the callbacks did. + sub_room.close().await?; + drop(sub_room); + + let released = timeout(Duration::from_secs(10), async { + while !remote_dropped() { + time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .is_ok(); + + drop(sub_events); + + assert!(released, "remote participant retained after the room was closed"); + Ok(()) +} + +/// A cancelled `close()` must still release the remote participants. +/// +/// `close` takes the room handle before it does anything else, so a caller that bounds it +/// with a timeout and drops the future cannot retry: a later call returns `AlreadyClosed`. +/// If the participant teardown sits behind the engine shutdown and the task joins, dropping +/// the future at any of those suspension points skips it permanently, and the publication +/// callbacks keep holding the participant and its `RtcEngine`. +#[cfg(feature = "__lk-e2e-test")] +#[test_log::test(tokio::test)] +async fn test_cancelled_close_releases_remote_participants() -> Result<()> { + let mut rooms = test_rooms(2).await?; + let (sub_room, mut sub_events) = rooms.remove(0); + let (pub_room, _pub_events) = rooms.remove(0); + + let mut solid_track = SolidColorTrack::new( + Arc::new(pub_room), + SolidColorParams { width: 320, height: 240, luma: 128 }, + ); + solid_track.publish(VideoCodec::VP8, false).await?; + + timeout(Duration::from_secs(15), async { + loop { + match sub_events.recv().await { + Some(RoomEvent::TrackSubscribed { .. }) => break Ok(()), + Some(_) => continue, + None => break Err(anyhow!("event stream ended before the track was subscribed")), + } + } + }) + .await??; + + let remote = sub_room + .remote_participants() + .into_values() + .next() + .ok_or_else(|| anyhow!("subscriber never saw the publisher"))?; + let remote_dropped = remote.drop_probe(); + drop(remote); + + // Poll the close exactly once, then drop it, cancelling at the earliest suspension point. + let cancelled = timeout(Duration::ZERO, sub_room.close()).await; + assert!(cancelled.is_err(), "close should have been cancelled at its first suspension point"); + drop(sub_room); + + let released = timeout(Duration::from_secs(10), async { + while !remote_dropped() { + time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .is_ok(); + + drop(sub_events); + + assert!(released, "a cancelled close left the remote participant retained"); + Ok(()) +}