diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index acdaf5bc0..e2821c574 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -485,6 +485,7 @@ func (p *ParticipantImpl) onPublisherAnswer(answer webrtc.SessionDescription) { prometheus.ServiceOperationCounter.WithLabelValues("answer", "error", "write_message").Add(1) return } + prometheus.ServiceOperationCounter.WithLabelValues("answer", "success", "").Add(1) p.TransportManager.PublisherLocalDescriptionSent() if p.isPublisher.Load() != p.CanPublish() { @@ -501,8 +502,6 @@ func (p *ParticipantImpl) onPublisherAnswer(answer webrtc.SessionDescription) { } } - prometheus.ServiceOperationCounter.WithLabelValues("answer", "success", "").Add(1) - if p.MigrateState() == types.MigrateStateSync { go p.handleMigrateMutedTrack() } @@ -1076,9 +1075,9 @@ func (p *ParticipantImpl) onSubscriberOffer(offer webrtc.SessionDescription) { }) if err != nil { prometheus.ServiceOperationCounter.WithLabelValues("offer", "error", "write_message").Add(1) - } else { - prometheus.ServiceOperationCounter.WithLabelValues("offer", "success", "").Add(1) + return } + prometheus.ServiceOperationCounter.WithLabelValues("offer", "success", "").Add(1) p.TransportManager.SubscriberLocalDescriptionSent() } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 12d6e53d6..260a25a11 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -36,7 +36,7 @@ const ( ReliableDataChannel = "_reliable" negotiationFrequency = 150 * time.Millisecond - negotiationFailedTimout = 15 * time.Second + negotiationFailedTimeout = 15 * time.Second dtlsRetransmissionInterval = 100 * time.Millisecond iceDisconnectedTimeout = 10 * time.Second // compatible for ice-lite with firefox client @@ -718,12 +718,7 @@ func (t *PCTransport) SendDataPacket(dp *livekit.DataPacket) error { } func (t *PCTransport) Close() { - t.lock.Lock() - if t.signalStateCheckTimer != nil { - t.signalStateCheckTimer.Stop() - t.signalStateCheckTimer = nil - } - t.lock.Unlock() + t.clearSignalStateCheckTimer() if t.streamAllocator != nil { t.streamAllocator.Stop() @@ -781,10 +776,7 @@ func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error { lastState := t.negotiationState t.negotiationState = negotiationStateNone - if t.signalStateCheckTimer != nil { - t.signalStateCheckTimer.Stop() - t.signalStateCheckTimer = nil - } + t.clearSignalStateCheckTimerLocked() for _, c := range t.pendingRemoteCandidates { if err := t.pc.AddICECandidate(c); err != nil { @@ -1018,6 +1010,37 @@ func (t *PCTransport) CreateAndSendAnswer(enableDTX bool) error { return nil } +func (t *PCTransport) clearSignalStateCheckTimer() { + t.lock.Lock() + t.clearSignalStateCheckTimerLocked() + t.lock.Unlock() +} + +func (t *PCTransport) clearSignalStateCheckTimerLocked() { + if t.signalStateCheckTimer != nil { + t.signalStateCheckTimer.Stop() + t.signalStateCheckTimer = nil + } +} + +func (t *PCTransport) setupSignalStateCheckTimerLocked() { + negotiateVersion := t.negotiateCounter.Inc() + t.clearSignalStateCheckTimerLocked() + t.signalStateCheckTimer = time.AfterFunc(negotiationFailedTimeout, func() { + t.lock.Lock() + t.clearSignalStateCheckTimerLocked() + + failed := t.negotiationState != negotiationStateNone + t.lock.Unlock() + + if t.negotiateCounter.Load() == negotiateVersion && failed { + if t.onNegotiationFailed != nil { + t.onNegotiationFailed() + } + } + }) +} + func (t *PCTransport) CreateAndSendOffer(options *webrtc.OfferOptions) error { t.lock.Lock() defer t.lock.Unlock() @@ -1136,21 +1159,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { t.restartAfterGathering = false t.negotiationPending = make(map[livekit.ParticipantID]bool) - negotiateVersion := t.negotiateCounter.Inc() - if t.signalStateCheckTimer != nil { - t.signalStateCheckTimer.Stop() - t.signalStateCheckTimer = nil - } - t.signalStateCheckTimer = time.AfterFunc(negotiationFailedTimout, func() { - t.lock.RLock() - failed := t.negotiationState != negotiationStateNone - t.lock.RUnlock() - if t.negotiateCounter.Load() == negotiateVersion && failed { - if t.onNegotiationFailed != nil { - t.onNegotiationFailed() - } - } - }) + t.setupSignalStateCheckTimerLocked() go t.onOffer(offer) return nil diff --git a/pkg/rtc/transport_test.go b/pkg/rtc/transport_test.go index 21d6b798d..9d5071294 100644 --- a/pkg/rtc/transport_test.go +++ b/pkg/rtc/transport_test.go @@ -268,7 +268,7 @@ func TestNegotiationFailed(t *testing.T) { transportA.CreateAndSendOffer(nil) require.Eventually(t, func() bool { return atomic.LoadInt32(&failed) == 1 - }, negotiationFailedTimout+time.Second, 10*time.Millisecond, "negotiation failed") + }, negotiationFailedTimeout+time.Second, 10*time.Millisecond, "negotiation failed") } func TestFilteringCandidates(t *testing.T) {