From 972c64a29b59fe5a5dd0751cc1013d80f8dae47a Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 20 Aug 2026 13:07:28 +0530 Subject: [PATCH] Check for ICE connection before closing participant on signal close. (#4780) * Check for ICE connection before closing participant on signal close. Only close the participant if ICE has not connected by the time signal source is closed. If ICE had connected, candidates have been exchanged and link was established. So, it should be resumable. Waiting for DTLS closed the participant sometimes in the windowa after ICE connection, but before DTLS finishes and that unnecessarily closed the participant forcing a full reconnect. * variable name --- pkg/rtc/participant.go | 8 ++- pkg/rtc/transport.go | 65 ++++++++++--------- pkg/rtc/transportmanager.go | 24 +++++-- pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 63 ++++++++++++++++++ test/client/client.go | 6 +- 6 files changed, 127 insertions(+), 40 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index f35228f33..8aab59858 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1013,7 +1013,7 @@ func (p *ParticipantImpl) OnClaimsChanged(callback func(types.LocalParticipant)) func (p *ParticipantImpl) HandleSignalSourceClose() { p.TransportManager.SetSignalSourceValid(false) - if !p.HasConnected() { + if !p.HasICEConnected() { _ = p.Close(false, types.ParticipantCloseReasonSignalSourceClose, false) } } @@ -1449,7 +1449,7 @@ func (p *ParticipantImpl) IsMigration() bool { } func (p *ParticipantImpl) recordRTCState(closeReason types.ParticipantCloseReason) { - if p.HasConnected() { + if p.HasICEConnected() { return } @@ -3199,6 +3199,10 @@ func (p *ParticipantImpl) GetPendingTrack(trackID livekit.TrackID) *livekit.Trac return nil } +func (p *ParticipantImpl) HasICEConnected() bool { + return p.TransportManager.HasSubscriberICEEverConnected() || p.TransportManager.HasPublisherICEEverConnected() +} + func (p *ParticipantImpl) HasConnected() bool { return p.TransportManager.HasSubscriberEverConnected() || p.TransportManager.HasPublisherEverConnected() } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index cdd3eb6e1..c15d2e1d4 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -223,14 +223,14 @@ type PCTransport struct { dataTrackDC *datachannel.DataChannelWriter[*webrtc.DataChannel] unlabeledDataChannels []*datachannel.DataChannelWriter[*webrtc.DataChannel] - iceStartedAt time.Time - iceConnectedAt time.Time - firstConnectedAt time.Time - connectedAt time.Time - tcpICETimer *time.Timer - connectAfterICETimer *time.Timer // timer to wait for pc to connect after ice connected - resetShortConnOnICERestart atomic.Bool - signalingRTT atomic.Uint32 // milliseconds + iceFirstStartedAt time.Time + iceFirstConnectedAt time.Time + peerConnectionFirstConnectedAt time.Time + peerConnectionLastconnectedAt time.Time + tcpICETimer *time.Timer + connectAfterICETimer *time.Timer // timer to wait for pc to connect after ice connected + resetShortConnOnICERestart atomic.Bool + signalingRTT atomic.Uint32 // milliseconds hasFullyEstablishedRecorded bool @@ -670,8 +670,8 @@ func (t *PCTransport) SetSignalingRTT(rtt uint32) { func (t *PCTransport) setICEStartedAt(at time.Time) { t.lock.Lock() - if t.iceStartedAt.IsZero() { - t.iceStartedAt = at + if t.iceFirstStartedAt.IsZero() { + t.iceFirstStartedAt = at // checklist of ice agent will be cleared on ice failed, get stats before that t.mayFailedICEStatsTimer = time.AfterFunc(iceFailedTimeoutTotal-time.Second, t.logMayFailedICEStats) @@ -702,15 +702,15 @@ func (t *PCTransport) setICEStartedAt(at time.Time) { func (t *PCTransport) setICEConnectedAt(at time.Time) { t.lock.Lock() - if t.iceConnectedAt.IsZero() { + if t.iceFirstConnectedAt.IsZero() { // // Record initial connection time. - // This prevents reset of connected at time if ICE goes `Connected` -> `Disconnected` -> `Connected`. + // This prevents reset of iceFirstConnectedAt if ICE goes `Connected` -> `Disconnected` -> `Connected`. // - t.iceConnectedAt = at + t.iceFirstConnectedAt = at // set failure timer for dtls handshake - iceDuration := at.Sub(t.iceStartedAt) + iceDuration := at.Sub(t.iceFirstStartedAt) connTimeoutAfterICE := min(max(minConnectTimeoutAfterICE, 3*iceDuration), maxConnectTimeoutAfterICE) t.params.Logger.Debugw("setting connection timer after ICE connected", "timeout", connTimeoutAfterICE, "iceDuration", iceDuration) t.connectAfterICETimer = time.AfterFunc(connTimeoutAfterICE, func() { @@ -777,9 +777,9 @@ func (t *PCTransport) logMayFailedICEStats() { func (t *PCTransport) resetShortConn() { t.params.Logger.Infow("resetting short connection on ICE restart") t.lock.Lock() - t.iceStartedAt = time.Time{} - t.iceConnectedAt = time.Time{} - t.connectedAt = time.Time{} + t.iceFirstStartedAt = time.Time{} + t.iceFirstConnectedAt = time.Time{} + t.peerConnectionLastconnectedAt = time.Time{} if t.connectAfterICETimer != nil { t.connectAfterICETimer.Stop() t.connectAfterICETimer = nil @@ -795,23 +795,23 @@ func (t *PCTransport) IsShortConnection(at time.Time) (bool, time.Duration) { t.lock.RLock() defer t.lock.RUnlock() - if t.iceConnectedAt.IsZero() { + if t.iceFirstConnectedAt.IsZero() { return false, 0 } - duration := at.Sub(t.iceConnectedAt) + duration := at.Sub(t.iceFirstConnectedAt) return duration < shortConnectionThreshold, duration } -func (t *PCTransport) setConnectedAt(at time.Time) bool { +func (t *PCTransport) setPeerConnectionConnectedAt(at time.Time) bool { t.lock.Lock() - t.connectedAt = at - if !t.firstConnectedAt.IsZero() { + t.peerConnectionLastconnectedAt = at + if !t.peerConnectionFirstConnectedAt.IsZero() { t.lock.Unlock() return false } - t.firstConnectedAt = at + t.peerConnectionFirstConnectedAt = at prometheus.RecordServiceOperationSuccess("peer_connection") prometheus.RecordPeerConnectionState(t.params.Transport, "connected") t.lock.Unlock() @@ -865,7 +865,7 @@ func (t *PCTransport) onPeerConnectionStateChange(state webrtc.PeerConnectionSta switch state { case webrtc.PeerConnectionStateConnected: t.clearConnTimer() - isInitialConnection := t.setConnectedAt(time.Now()) + isInitialConnection := t.setPeerConnectionConnectedAt(time.Now()) if isInitialConnection { t.params.Handler.OnInitialConnected() @@ -993,7 +993,7 @@ func (t *PCTransport) isFullyEstablished() bool { dataChannelReady := t.params.UseOneShotSignallingMode || t.firstOfferNoDataChannel || (t.reliableDCOpened && t.lossyDCOpened) - return dataChannelReady && !t.connectedAt.IsZero() + return dataChannelReady && !t.peerConnectionLastconnectedAt.IsZero() } func (t *PCTransport) SetPreferTCP(preferTCP bool) { @@ -1440,18 +1440,25 @@ func (t *PCTransport) IsEstablished() bool { return t.pc.ConnectionState() != webrtc.PeerConnectionStateNew } -func (t *PCTransport) HasEverConnected() bool { +func (t *PCTransport) ICEHasEverConnected() bool { t.lock.RLock() defer t.lock.RUnlock() - return !t.firstConnectedAt.IsZero() + return !t.iceFirstConnectedAt.IsZero() } -func (t *PCTransport) FirstConnectedAt() time.Time { +func (t *PCTransport) PeerConnectionHasEverConnected() bool { t.lock.RLock() defer t.lock.RUnlock() - return t.firstConnectedAt + return !t.peerConnectionFirstConnectedAt.IsZero() +} + +func (t *PCTransport) PeerConnectionFirstConnectedAt() time.Time { + t.lock.RLock() + defer t.lock.RUnlock() + + return t.peerConnectionFirstConnectedAt } func (t *PCTransport) GetICEConnectionInfo() *types.ICEConnectionInfo { diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 9e8cf51b1..108651f21 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -232,12 +232,16 @@ func (t *TransportManager) SubscriberClose() { t.subscriber.Close() } +func (t *TransportManager) HasPublisherICEEverConnected() bool { + return t.publisher.ICEHasEverConnected() +} + func (t *TransportManager) HasPublisherEverConnected() bool { - return t.publisher.HasEverConnected() + return t.publisher.PeerConnectionHasEverConnected() } func (t *TransportManager) PublisherFirstConnectedAt() time.Time { - return t.publisher.FirstConnectedAt() + return t.publisher.PeerConnectionFirstConnectedAt() } func (t *TransportManager) IsPublisherEstablished() bool { @@ -272,19 +276,27 @@ func (t *TransportManager) GetSubscriberRTT() (float64, bool) { } } +func (t *TransportManager) HasSubscriberICEEverConnected() bool { + if t.params.UseOneShotSignallingMode || t.params.UseSinglePeerConnection { + return t.publisher.ICEHasEverConnected() + } else { + return t.subscriber.ICEHasEverConnected() + } +} + func (t *TransportManager) HasSubscriberEverConnected() bool { if t.params.UseOneShotSignallingMode || t.params.UseSinglePeerConnection { - return t.publisher.HasEverConnected() + return t.publisher.PeerConnectionHasEverConnected() } else { - return t.subscriber.HasEverConnected() + return t.subscriber.PeerConnectionHasEverConnected() } } func (t *TransportManager) SubscriberFirstConnectedAt() time.Time { if t.params.UseOneShotSignallingMode || t.params.UseSinglePeerConnection { - return t.publisher.FirstConnectedAt() + return t.publisher.PeerConnectionFirstConnectedAt() } else { - return t.subscriber.FirstConnectedAt() + return t.subscriber.PeerConnectionFirstConnectedAt() } } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index ddc462e1f..9c3a23db3 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -425,6 +425,7 @@ type LocalParticipant interface { GetPlayoutDelayConfig() *livekit.PlayoutDelay GetPendingTrack(trackID livekit.TrackID) *livekit.TrackInfo GetICEConnectionInfo() []*ICEConnectionInfo + HasICEConnected() bool HasConnected() bool GetEnabledPublishCodecs() []*livekit.Codec GetPublisherICESessionUfrag() (string, error) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index d4fcfd21b..e0900c54f 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -774,6 +774,16 @@ type FakeLocalParticipant struct { hasConnectedReturnsOnCall map[int]struct { result1 bool } + HasICEConnectedStub func() bool + hasICEConnectedMutex sync.RWMutex + hasICEConnectedArgsForCall []struct { + } + hasICEConnectedReturns struct { + result1 bool + } + hasICEConnectedReturnsOnCall map[int]struct { + result1 bool + } HasPermissionStub func(livekit.TrackID, livekit.ParticipantIdentity) bool hasPermissionMutex sync.RWMutex hasPermissionArgsForCall []struct { @@ -5618,6 +5628,59 @@ func (fake *FakeLocalParticipant) HasConnectedReturnsOnCall(i int, result1 bool) }{result1} } +func (fake *FakeLocalParticipant) HasICEConnected() bool { + fake.hasICEConnectedMutex.Lock() + ret, specificReturn := fake.hasICEConnectedReturnsOnCall[len(fake.hasICEConnectedArgsForCall)] + fake.hasICEConnectedArgsForCall = append(fake.hasICEConnectedArgsForCall, struct { + }{}) + stub := fake.HasICEConnectedStub + fakeReturns := fake.hasICEConnectedReturns + fake.recordInvocation("HasICEConnected", []interface{}{}) + fake.hasICEConnectedMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) HasICEConnectedCallCount() int { + fake.hasICEConnectedMutex.RLock() + defer fake.hasICEConnectedMutex.RUnlock() + return len(fake.hasICEConnectedArgsForCall) +} + +func (fake *FakeLocalParticipant) HasICEConnectedCalls(stub func() bool) { + fake.hasICEConnectedMutex.Lock() + defer fake.hasICEConnectedMutex.Unlock() + fake.HasICEConnectedStub = stub +} + +func (fake *FakeLocalParticipant) HasICEConnectedReturns(result1 bool) { + fake.hasICEConnectedMutex.Lock() + defer fake.hasICEConnectedMutex.Unlock() + fake.HasICEConnectedStub = nil + fake.hasICEConnectedReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) HasICEConnectedReturnsOnCall(i int, result1 bool) { + fake.hasICEConnectedMutex.Lock() + defer fake.hasICEConnectedMutex.Unlock() + fake.HasICEConnectedStub = nil + if fake.hasICEConnectedReturnsOnCall == nil { + fake.hasICEConnectedReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.hasICEConnectedReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) HasPermission(arg1 livekit.TrackID, arg2 livekit.ParticipantIdentity) bool { fake.hasPermissionMutex.Lock() ret, specificReturn := fake.hasPermissionReturnsOnCall[len(fake.hasPermissionArgsForCall)] diff --git a/test/client/client.go b/test/client/client.go index bb38cdaf3..bd8c76f20 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -867,9 +867,9 @@ func (c *RTCClient) SetAttributes(attrs map[string]string) error { func (c *RTCClient) hasPrimaryEverConnected() bool { if c.subscriberAsPrimary.Load() { - return c.subscriber.HasEverConnected() + return c.subscriber.PeerConnectionHasEverConnected() } else { - return c.publisher.HasEverConnected() + return c.publisher.PeerConnectionHasEverConnected() } } @@ -1113,7 +1113,7 @@ func (c *RTCClient) ensurePublisherConnected() error { return c.ctx.Err() } - if c.publisher.HasEverConnected() { + if c.publisher.PeerConnectionHasEverConnected() { return nil }