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

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

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 {

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

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)

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)]

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
 	}