diff --git a/.changeset/unpublish_cleanup_when_transport_gone.md b/.changeset/unpublish_cleanup_when_transport_gone.md new file mode 100644 index 000000000..f082427c4 --- /dev/null +++ b/.changeset/unpublish_cleanup_when_transport_gone.md @@ -0,0 +1,16 @@ +--- +livekit: patch +livekit-capture: patch +livekit-ffi: patch +--- + +Fix a local publication and its track leaking when unpublish cannot reach the transport. + +Unpublishing asks the engine to remove the RTP sender and stopped on failure, skipping the +cleanup that detaches the track from its publication. The publication holds its track and +the track holds the publication back through its mute callbacks, so the pair kept itself +alive along with the transceiver and its peer connection. Removing the sender fails on two +routine paths — an abnormal disconnect, where the transport is already closed, and a full +reconnect, where the sender belongs to the replaced transport — and the error was discarded, +so neither surfaced. The local cleanup now always runs and the failure is reported once +local state is consistent. diff --git a/livekit/src/room/participant/local_participant.rs b/livekit/src/room/participant/local_participant.rs index b4545fc1b..49efadd46 100644 --- a/livekit/src/room/participant/local_participant.rs +++ b/livekit/src/room/participant/local_participant.rs @@ -667,16 +667,28 @@ impl LocalParticipant { let track = publication.track().unwrap(); let sender = track.transceiver().unwrap().sender(); - self.inner.rtc_engine.remove_track(sender)?; + // Removing the sender fails whenever the publisher transport is no longer the + // one this sender belongs to: an abnormal disconnect closes the transport + // before teardown runs, and a full reconnect replaces it outright. The local + // publication still has to be torn down in both cases, so the failure must not + // short-circuit the cleanup below — `set_track(None)` is the only thing that + // unregisters the track's mute callbacks, and without it the publication and + // its track hold each other, keeping the transceiver and its peer connection + // alive with them. Report the failure once the local state is consistent. + let removed = self.inner.rtc_engine.remove_track(sender); track.set_transceiver(None); - if let Some(local_track_unpublished) = - self.local.events.local_track_unpublished.lock().as_ref() - { - local_track_unpublished(self.clone(), publication.clone()); + if removed.is_ok() { + if let Some(local_track_unpublished) = + self.local.events.local_track_unpublished.lock().as_ref() + { + local_track_unpublished(self.clone(), publication.clone()); + } } publication.set_track(None); + + removed?; self.inner.rtc_engine.publisher_negotiation_needed(); Ok(publication) diff --git a/livekit/src/room/publication/local.rs b/livekit/src/room/publication/local.rs index dcf543481..7470b5fe1 100644 --- a/livekit/src/room/publication/local.rs +++ b/livekit/src/room/publication/local.rs @@ -49,6 +49,15 @@ impl LocalTrackPublication { } } + /// Test-only: returns a probe that reports whether this publication's internals have + /// been dropped. The publication holds its track and the track holds the publication + /// back through the mute callbacks, so only unregistering them releases either. + #[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) fn on_muted(&self, f: impl Fn(TrackPublication) + Send + 'static) { *self.inner.events.muted.lock() = Some(Box::new(f)); } diff --git a/livekit/tests/room_test.rs b/livekit/tests/room_test.rs index 5e05f9246..5beff7300 100644 --- a/livekit/tests/room_test.rs +++ b/livekit/tests/room_test.rs @@ -14,12 +14,16 @@ #[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}, + livekit_api::services::room::RoomClient, + std::{env, sync::Arc, time::Duration}, tokio::time::{self, timeout}, }; @@ -140,3 +144,76 @@ async fn test_close_releases_room_session() -> Result<()> { assert!(session_dropped(), "room callbacks retained the room session after close"); Ok(()) } + +/// An unpublish that cannot reach the transport must still complete its local cleanup. +/// +/// `unpublish_track` removes the publication from the participant, then asks the engine to +/// remove the RTP sender with `?`. On an abnormal disconnect the publisher transport has +/// already been closed by the time teardown runs, so that call fails and `?` skips +/// everything after it — including `publication.set_track(None)`, the only thing that +/// unregisters the track's mute callbacks. Those callbacks hold the publication while the +/// publication holds the track, so the pair keeps itself alive along with the transceiver +/// and the peer connection behind it. +/// +/// The same failure happens on every full reconnect, where the sender belongs to the +/// previous session. `close()` discards the error, so nothing ever surfaces it. +#[cfg(feature = "__lk-e2e-test")] +#[test_log::test(tokio::test)] +async fn test_unpublish_cleans_up_when_transport_is_gone() -> Result<()> { + let (room, mut events) = test_rooms(1).await?.pop().unwrap(); + let room_name = room.name(); + let room = Arc::new(room); + + let mut solid_track = + SolidColorTrack::new(room.clone(), SolidColorParams { width: 320, height: 240, luma: 128 }); + solid_track.publish(VideoCodec::VP8, false).await?; + + let publication = room + .local_participant() + .track_publications() + .into_values() + .next() + .ok_or_else(|| anyhow!("the track was never published"))?; + let publication_dropped = publication.drop_probe(); + drop(publication); + + // Delete the room server-side. The engine closes its transports before reporting + // Disconnected, so the room's teardown runs against an already-closed publisher — + // which is exactly when removing the sender fails. + let api_key = env::var("LIVEKIT_API_KEY").unwrap_or_else(|_| "devkey".into()); + let api_secret = env::var("LIVEKIT_API_SECRET").unwrap_or_else(|_| "secret".into()); + let server_url = env::var("LIVEKIT_URL").unwrap_or_else(|_| "ws://localhost:7880".into()); + let http_url = server_url.replacen("ws", "http", 1); + RoomClient::with_api_key(&http_url, &api_key, &api_secret).delete_room(&room_name).await?; + + timeout(Duration::from_secs(15), async { + loop { + match events.recv().await { + Some(RoomEvent::Disconnected { .. }) => break Ok(()), + Some(_) => continue, + None => break Err(anyhow!("event stream ended before the room disconnected")), + } + } + }) + .await??; + + // Drop every reference held outside the SDK, so anything still alive is held only by + // the publication <-> track cycle. + drop(events); + drop(solid_track); + drop(room); + + let released = timeout(Duration::from_secs(10), async { + while !publication_dropped() { + time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .is_ok(); + + assert!( + released, + "local publication retained after an unpublish that could not reach the transport" + ); + Ok(()) +}