Close peer connections before awaiting signal teardown

.changeset/close_peer_connections_before_signal_teardown.md

@@ -0,0 +1,12 @@
+---
+livekit: patch
+livekit-ffi: patch
+---
+
+# Close peer connections before awaiting signal teardown
+
+`SessionInner::close` released the peer connections only after two awaits that can block
+indefinitely, so cancelling `close()` — for example by wrapping it in a timeout — left the
+transports open and their ICE UDP sockets bound for the lifetime of the process. Long-lived
+clients eventually exhausted their file descriptors. The transports are now closed before
+the first await, which makes the teardown safe to cancel.

livekit/src/room/mod.rs

@@ -908,6 +908,14 @@ impl Room {
         self.inner.rtc_engine.drop_disconnected_updates(enabled);
     }
 
+    /// Test-only: the publisher transport's current connection state. Lets a test assert
+    /// that teardown really closed the transport, rather than inferring it from room-level
+    /// state that can reach `Disconnected` while the transport is still open.
+    #[cfg(feature = "__lk-e2e-test")]
+    pub fn publisher_connection_state(&self) -> libwebrtc::prelude::PeerConnectionState {
+        self.inner.rtc_engine.session().publisher_connection_state()
+    }
+
     pub async fn get_stats(&self) -> EngineResult<SessionStats> {
         self.inner.rtc_engine.get_stats().await
     }

livekit/src/rtc_engine/rtc_session.rs

@@ -845,6 +845,13 @@ impl RtcSession {
         self.inner.drop_disconnected_updates.store(enabled, Ordering::Release);
     }
 
+    /// Test-only: the publisher transport's current connection state, so tests can assert
+    /// that teardown actually closed it rather than inferring it from room-level state.
+    #[cfg(feature = "__lk-e2e-test")]
+    pub fn publisher_connection_state(&self) -> PeerConnectionState {
+        self.inner.publisher_pc.peer_connection().connection_state()
+    }
+
     pub async fn wait_pc_connection(&self) -> EngineResult<()> {
         self.inner.wait_pc_connection().await
     }
@@ -1981,6 +1988,15 @@ impl SessionInner {
         self.closed.store(true, Ordering::Release);
         self.pc_state_notify.notify_waiters();
 
+        // Both awaits below are unbounded, and `PeerTransport::close` is synchronous, so
+        // closing here — before the future can suspend — is what stops a cancelled
+        // `close()` leaving the ICE sockets bound for the process's lifetime. The
+        // signalling socket is unaffected, so the Leave below still goes out.
+        self.publisher_pc.close();
+        if let Some(ref sub_pc) = self.subscriber_pc {
+            sub_pc.close();
+        }
+
         self.signal_client
             .send(proto::signal_request::Message::Leave(proto::LeaveRequest {
                 action: proto::leave_request::Action::Disconnect.into(),
@@ -1990,10 +2006,6 @@ impl SessionInner {
             .await;
 
         self.signal_client.close().await;
-        self.publisher_pc.close();
-        if let Some(ref sub_pc) = self.subscriber_pc {
-            sub_pc.close();
-        }
     }
 
     async fn simulate_scenario(self: &Arc<Self>, scenario: SimulateScenario) -> EngineResult<()> {

livekit/tests/room_test.rs

@@ -17,6 +17,7 @@ use {
     anyhow::{Ok, Result},
     chrono::{TimeDelta, TimeZone, Utc},
     common::test_rooms,
+    libwebrtc::prelude::PeerConnectionState,
     livekit::{ConnectionState, ParticipantKind, RoomEvent},
     std::time::Duration,
     tokio::time::{self, timeout},
@@ -89,3 +90,35 @@ async fn test_participant_disconnect() -> Result<()> {
     timeout(Duration::from_secs(15), wait_for_disconnected).await??;
     Ok(())
 }
+
+/// A cancelled `close()` must still have closed the peer connections.
+///
+/// `SessionInner::close` ends with two unbounded awaits, so a caller bounding it with a
+/// timeout used to drop the future before the transports were closed — and nothing else
+/// closes them, so their ICE sockets stayed bound for the process's lifetime. Cancelling at
+/// the first `.await` makes this deterministic: the assertion holds only if the close ran
+/// before the future suspended.
+#[cfg(feature = "__lk-e2e-test")]
+#[test_log::test(tokio::test)]
+async fn test_cancelled_close_still_closes_peer_connections() -> Result<()> {
+    let (room, _) = test_rooms(1).await?.pop().unwrap();
+    assert_eq!(
+        room.publisher_connection_state(),
+        PeerConnectionState::Connected,
+        "publisher should be connected before teardown, or the assertion below proves nothing"
+    );
+
+    // Poll the close exactly once, then drop it. `Duration::ZERO` elapses on the first poll,
+    // so `timeout` yields `Err` the moment the inner future suspends — cancelling it at the
+    // earliest possible point rather than at some arbitrary later one.
+    let cancelled = timeout(Duration::ZERO, room.close()).await;
+    assert!(cancelled.is_err(), "close should have been cancelled at its first suspension point");
+
+    assert_eq!(
+        room.publisher_connection_state(),
+        PeerConnectionState::Closed,
+        "a cancelled close must not leave the publisher transport open: its ICE sockets are \
+         only released by PeerConnection::close, so anything else leaks them permanently"
+    );
+    Ok(())
+}