fix(room): finish local unpublish cleanup when the transport is gone

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

livekit/src/room/participant/local_participant.rs

@@ -387,7 +387,7 @@
             r#type: proto::TrackType::from(track.kind()) as i32,
             muted: track.is_muted(),
             source: proto::TrackSource::from(options.source) as i32,
             disable_dtx: !options.dtx,
             disable_red,
             encryption: proto::encryption::Type::from(self.local.encryption_type) as i32,
             stream: options.stream.clone(),
@@ -667,16 +667,28 @@
             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)

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));
     }

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