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 }