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