Improve connectivity check (#1366)

* Add Timer to detect dtls failure quickly

* Fix pc state check in timeout after ice

* More strict conditions to switch candidate type

* log for signal interuppt

* typo

pkg/rtc/signalhandler.go

@@ -8,6 +8,8 @@ import (
 )
 
 func HandleParticipantSignal(room types.Room, participant types.LocalParticipant, req *livekit.SignalRequest, pLogger logger.Logger) error {
+	participant.UpdateLastSeenSignal()
+
 	switch msg := req.Message.(type) {
 	case *livekit.SignalRequest_Offer:
 		participant.HandleOffer(FromProtoSessionDescription(msg.Offer))

pkg/rtc/transport.go

@@ -39,9 +39,11 @@ const (
 	negotiationFailedTimeout   = 15 * time.Second
 	dtlsRetransmissionInterval = 100 * time.Millisecond
 
-	iceDisconnectedTimeout = 10 * time.Second // compatible for ice-lite with firefox client
-	iceFailedTimeout       = 25 * time.Second // pion's default
-	iceKeepaliveInterval   = 2 * time.Second  // pion's default
+	iceDisconnectedTimeout    = 10 * time.Second // compatible for ice-lite with firefox client
+	iceFailedTimeout          = 25 * time.Second // pion's default
+	iceKeepaliveInterval      = 2 * time.Second  // pion's default
+	minConnectTimeoutAfterICE = 5 * time.Second  // min duration for waiting pc to connect after ICE is connected
+	maxConnectTimeoutAfterICE = 20 * time.Second // max duration for waiting pc to connect after ICE is connected
 
 	shortConnectionThreshold = 90 * time.Second
 )
@@ -153,8 +155,10 @@ type PCTransport struct {
 	lossyDCOpened    bool
 	onDataPacket     func(kind livekit.DataPacket_Kind, data []byte)
 
-	iceConnectedAt time.Time
-	connectedAt    time.Time
+	iceStartedAt         time.Time
+	iceConnectedAt       time.Time
+	connectedAt          time.Time
+	connectAfterICETimer *time.Timer // timer to wait for pc to connect after ice connected
 
 	onFullyEstablished func()
 
@@ -397,6 +401,14 @@ func (t *PCTransport) createPeerConnection() error {
 	return nil
 }
 
+func (t *PCTransport) setICEStartedAt(at time.Time) {
+	t.lock.Lock()
+	if t.iceStartedAt.IsZero() {
+		t.iceStartedAt = at
+	}
+	t.lock.Unlock()
+}
+
 func (t *PCTransport) setICEConnectedAt(at time.Time) {
 	t.lock.Lock()
 	if t.iceConnectedAt.IsZero() {
@@ -405,6 +417,24 @@ func (t *PCTransport) setICEConnectedAt(at time.Time) {
 		// This prevents reset of connected at time if ICE goes `Connected` -> `Disconnected` -> `Connected`.
 		//
 		t.iceConnectedAt = at
+
+		// set failure timer for dtls handshake
+		iceCost := at.Sub(t.iceStartedAt)
+		connTimeoutAfterICE := 3 * iceCost
+		if connTimeoutAfterICE < minConnectTimeoutAfterICE {
+			connTimeoutAfterICE = minConnectTimeoutAfterICE
+		} else if connTimeoutAfterICE > maxConnectTimeoutAfterICE {
+			connTimeoutAfterICE = maxConnectTimeoutAfterICE
+		}
+		t.params.Logger.Debugw("setting connection timer after ice connected", "timeout", connTimeoutAfterICE, "iceCost", iceCost)
+		t.connectAfterICETimer = time.AfterFunc(connTimeoutAfterICE, func() {
+			state := t.pc.ConnectionState()
+			// if pc is still checking or connected but not fully established after timeout, then fire connection fail
+			if state != webrtc.PeerConnectionStateClosed && state != webrtc.PeerConnectionStateFailed && !t.isFullyEstablished() {
+				t.params.Logger.Infow("connect timeout after ICE connected", "timeout", connTimeoutAfterICE, "iceCost", iceCost)
+				t.handleConnectionFailed()
+			}
+		})
 	}
 	t.lock.Unlock()
 }
@@ -503,6 +533,9 @@ func (t *PCTransport) onICEConnectionStateChange(state webrtc.ICEConnectionState
 		} else {
 			t.params.Logger.Infow("selected ICE candidate pair", "pair", pair)
 		}
+
+	case webrtc.ICEConnectionStateChecking:
+		t.setICEStartedAt(time.Now())
 	}
 }
 
@@ -510,6 +543,7 @@ func (t *PCTransport) onPeerConnectionStateChange(state webrtc.PeerConnectionSta
 	t.params.Logger.Debugw("peer connection state change", "state", state.String())
 	switch state {
 	case webrtc.PeerConnectionStateConnected:
+		t.clearConnTimerAfterICE()
 		isInitialConnection := t.setConnectedAt(time.Now())
 		if isInitialConnection {
 			if onInitialConnected := t.getOnInitialConnected(); onInitialConnected != nil {
@@ -520,6 +554,7 @@ func (t *PCTransport) onPeerConnectionStateChange(state webrtc.PeerConnectionSta
 		}
 	case webrtc.PeerConnectionStateFailed:
 		t.params.Logger.Infow("peer connection failed")
+		t.clearConnTimerAfterICE()
 		t.logICECandidates()
 		t.handleConnectionFailed()
 	}
@@ -559,17 +594,20 @@ func (t *PCTransport) onDataChannel(dc *webrtc.DataChannel) {
 }
 
 func (t *PCTransport) maybeNotifyFullyEstablished() {
-	t.lock.RLock()
-	fullyEstablished := t.reliableDCOpened && t.lossyDCOpened && !t.connectedAt.IsZero()
-	t.lock.RUnlock()
-
-	if fullyEstablished {
+	if t.isFullyEstablished() {
 		if onFullyEstablished := t.getOnFullyEstablished(); onFullyEstablished != nil {
 			onFullyEstablished()
 		}
 	}
 }
 
+func (t *PCTransport) isFullyEstablished() bool {
+	t.lock.RLock()
+	fullyEstablished := t.reliableDCOpened && t.lossyDCOpened && !t.connectedAt.IsZero()
+	t.lock.RUnlock()
+	return fullyEstablished
+}
+
 func (t *PCTransport) SetPreferTCP(preferTCP bool) {
 	t.preferTCP.Store(preferTCP)
 }
@@ -782,6 +820,17 @@ func (t *PCTransport) Close() {
 	}
 
 	_ = t.pc.Close()
+
+	t.clearConnTimerAfterICE()
+}
+
+func (t *PCTransport) clearConnTimerAfterICE() {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	if t.connectAfterICETimer != nil {
+		t.connectAfterICETimer.Stop()
+		t.connectAfterICETimer = nil
+	}
 }
 
 func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription) {

pkg/rtc/transportmanager.go

@@ -4,6 +4,7 @@ import (
 	"math/bits"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/pion/rtcp"
 	"github.com/pion/sdp/v3"
@@ -21,7 +22,8 @@ import (
 )
 
 const (
-	failureCountThreshold = 2
+	failureCountThreshold     = 2
+	preferNextByFailureWindow = time.Minute
 
 	// when RR report loss percentage over this threshold, we consider it is a unstable event
 	udpLossFracUnstable = 25
@@ -57,6 +59,8 @@ type TransportManager struct {
 	subscriber              *PCTransport
 	failureCount            int
 	isTransportReconfigured bool
+	lastFailure             time.Time
+	lastSignalAt            time.Time
 
 	pendingOfferPublisher        *webrtc.SessionDescription
 	pendingDataChannelsPublisher []*livekit.DataChannelInfo
@@ -460,6 +464,7 @@ func (t *TransportManager) configureICE(iceConfig *livekit.ICEConfig, reset bool
 		t.failureCount = 0
 		t.isTransportReconfigured = !reset
 		t.udpLossUnstableCount = 0
+		t.lastFailure = time.Time{}
 	}
 
 	if isEqual {
@@ -512,6 +517,18 @@ func (t *TransportManager) handleConnectionFailed(isShortLived bool) {
 		return
 	}
 
+	lastSignalSince := time.Since(t.lastSignalAt)
+	if lastSignalSince > iceFailedTimeout {
+		// the failed might cause by network interrupt because we have not seen any signal in the time window too
+		// so don't switch to next candidate type
+		t.params.Logger.Infow("ignoring prefer candidate check by ICE failure because no signal received in the ice failed window",
+			"lastSignalSince", lastSignalSince)
+		t.failureCount = 0
+		t.lastFailure = time.Time{}
+		t.lock.Unlock()
+		return
+	}
+
 	//
 	// Checking only `PreferenceSubcriber` field although any connection failure (PUBLISHER OR SUBSCRIBER) will
 	// flow through here.
@@ -533,7 +550,9 @@ func (t *TransportManager) handleConnectionFailed(isShortLived bool) {
 		preferNext = getNext(t.iceConfig)
 	} else {
 		t.failureCount++
-		if t.failureCount < failureCountThreshold {
+		lastFailure := t.lastFailure
+		t.lastFailure = time.Now()
+		if t.failureCount < failureCountThreshold || time.Since(lastFailure) > preferNextByFailureWindow {
 			t.lock.Unlock()
 			return
 		}
@@ -668,5 +687,15 @@ func (t *TransportManager) UpdateRTT(rtt uint32, isUDP bool) {
 		}
 	} else {
 		t.tcpRTT = rtt
+
+		// TODO: considering using tcp rtt to calculate ice connection cost, if ice connection can't be established
+		// within 5 * tcp rtt(at least 5s), means udp traffic might be block/dropped, switch to tcp.
+		// Currently, most cases reported is that ice connected but subsequent connection, so left the thinking for now.
 	}
 }
+
+func (t *TransportManager) UpdateLastSeenSignal() {
+	t.lock.Lock()
+	t.lastSignalAt = time.Now()
+	t.lock.Unlock()
+}

pkg/rtc/types/interfaces.go

@@ -245,6 +245,7 @@ type LocalParticipant interface {
 
 	SetResponseSink(sink routing.MessageSink)
 	CloseSignalConnection()
+	UpdateLastSeenSignal()
 
 	// permissions
 	ClaimGrants() *auth.ClaimGrants

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -713,6 +713,10 @@ type FakeLocalParticipant struct {
 	unsubscribeFromTrackArgsForCall []struct {
 		arg1 livekit.TrackID
 	}
+	UpdateLastSeenSignalStub        func()
+	updateLastSeenSignalMutex       sync.RWMutex
+	updateLastSeenSignalArgsForCall []struct {
+	}
 	UpdateMediaLossStub        func(livekit.NodeID, livekit.TrackID, uint32) error
 	updateMediaLossMutex       sync.RWMutex
 	updateMediaLossArgsForCall []struct {
@@ -4580,6 +4584,30 @@ func (fake *FakeLocalParticipant) UnsubscribeFromTrackArgsForCall(i int) livekit
 	return argsForCall.arg1
 }
 
+func (fake *FakeLocalParticipant) UpdateLastSeenSignal() {
+	fake.updateLastSeenSignalMutex.Lock()
+	fake.updateLastSeenSignalArgsForCall = append(fake.updateLastSeenSignalArgsForCall, struct {
+	}{})
+	stub := fake.UpdateLastSeenSignalStub
+	fake.recordInvocation("UpdateLastSeenSignal", []interface{}{})
+	fake.updateLastSeenSignalMutex.Unlock()
+	if stub != nil {
+		fake.UpdateLastSeenSignalStub()
+	}
+}
+
+func (fake *FakeLocalParticipant) UpdateLastSeenSignalCallCount() int {
+	fake.updateLastSeenSignalMutex.RLock()
+	defer fake.updateLastSeenSignalMutex.RUnlock()
+	return len(fake.updateLastSeenSignalArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) UpdateLastSeenSignalCalls(stub func()) {
+	fake.updateLastSeenSignalMutex.Lock()
+	defer fake.updateLastSeenSignalMutex.Unlock()
+	fake.UpdateLastSeenSignalStub = stub
+}
+
 func (fake *FakeLocalParticipant) UpdateMediaLoss(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 uint32) error {
 	fake.updateMediaLossMutex.Lock()
 	ret, specificReturn := fake.updateMediaLossReturnsOnCall[len(fake.updateMediaLossArgsForCall)]
@@ -5158,6 +5186,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.uncacheDownTrackMutex.RUnlock()
 	fake.unsubscribeFromTrackMutex.RLock()
 	defer fake.unsubscribeFromTrackMutex.RUnlock()
+	fake.updateLastSeenSignalMutex.RLock()
+	defer fake.updateLastSeenSignalMutex.RUnlock()
 	fake.updateMediaLossMutex.RLock()
 	defer fake.updateMediaLossMutex.RUnlock()
 	fake.updateRTTMutex.RLock()

pkg/service/rtcservice.go

@@ -333,7 +333,6 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			if perr == nil && signalStats != nil {
 				signalStats.AddBytes(uint64(count), true)
 			}
-			continue
 		}
 
 		switch m := req.Message.(type) {