fix(room): release remote participants when the room is closed

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

livekit/src/room/mod.rs

@@ -1179,9 +1179,40 @@
         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<RtcEngine>` 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 @@
         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(())
     }
@@ -1938,7 +1972,7 @@
         // Back-compat raw-header event (non-internal topics only). The header topic alone
         // determines internal-ness, so it's gated here without consulting the actor.
         if !is_internal_topic(&header.topic) {
             let event = RoomEvent::StreamHeaderReceived {
                 header: header.clone(),
                 participant_identity: participant_identity.clone(),
             };
@@ -2463,12 +2497,12 @@
                 // topic of the stream they belong to for the internal check below.
                 ds::incoming::OutputEvent::ChunkReceived(ds::incoming::ChunkReceived { chunk, participant_identity, topic }) => {
                     if !topic.as_deref().is_some_and(is_internal_topic) {
                         dispatcher.dispatch(&RoomEvent::StreamChunkReceived { chunk: chunk.into(), participant_identity: participant_identity.into() });
                     }
                 }
                 ds::incoming::OutputEvent::TrailerReceived(ds::incoming::TrailerReceived { trailer, participant_identity, topic }) => {
                     if !topic.as_deref().is_some_and(is_internal_topic) {
                         dispatcher.dispatch(&RoomEvent::StreamTrailerReceived { trailer: trailer.into(), participant_identity: participant_identity.into() });
                     }
                 }
                 // The Rust SDK observes completion through the reader itself; the explicit

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,

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