diff --git a/.changeset/close_peer_connections_before_signal_teardown.md b/.changeset/close_peer_connections_before_signal_teardown.md new file mode 100644 index 000000000..069ce1392 --- /dev/null +++ b/.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. diff --git a/livekit/src/room/mod.rs b/livekit/src/room/mod.rs index b02d88e54..e5089f4fb 100644 --- a/livekit/src/room/mod.rs +++ b/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 { self.inner.rtc_engine.get_stats().await } diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index bb6c45f38..c225a6a77 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/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, scenario: SimulateScenario) -> EngineResult<()> { diff --git a/livekit/tests/room_test.rs b/livekit/tests/room_test.rs index b81ce1ae4..11479c27c 100644 --- a/livekit/tests/room_test.rs +++ b/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(()) +}