Adding transport fallback (#866)

* Adding transport fallback

Commented out for now so that we can gather some data.

* Promoting a few logs to info

pkg/rtc/participant.go

@@ -277,8 +277,17 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
 	} else {
 		p.activeCounter.Add(2)
 	}
+
+	primaryPC.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
+		p.handleICEStateChange(true, state)
+	})
+	secondaryPC.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
+		p.handleICEStateChange(false, state)
+	})
+
 	primaryPC.OnConnectionStateChange(p.handlePrimaryStateChange)
 	secondaryPC.OnConnectionStateChange(p.handleSecondaryStateChange)
+
 	p.publisher.pc.OnTrack(p.onMediaTrack)
 	p.publisher.pc.OnDataChannel(p.onDataChannel)
 
@@ -989,7 +998,7 @@ func (p *ParticipantImpl) UpdateSubscribedTrackSettings(trackID livekit.TrackID,
 
 // AddSubscribedTrack adds a track to the participant's subscribed list
 func (p *ParticipantImpl) AddSubscribedTrack(subTrack types.SubscribedTrack) {
-	p.params.Logger.Debugw("added subscribedTrack",
+	p.params.Logger.Infow("added subscribedTrack",
 		"publisherID", subTrack.PublisherID(),
 		"publisherIdentity", subTrack.PublisherIdentity(),
 		"trackID", subTrack.ID())
@@ -1031,7 +1040,7 @@ func (p *ParticipantImpl) AddSubscribedTrack(subTrack types.SubscribedTrack) {
 
 // RemoveSubscribedTrack removes a track to the participant's subscribed list
 func (p *ParticipantImpl) RemoveSubscribedTrack(subTrack types.SubscribedTrack) {
-	p.params.Logger.Debugw("removed subscribedTrack",
+	p.params.Logger.Infow("removed subscribedTrack",
 		"publisherID", subTrack.PublisherID(),
 		"publisherIdentity", subTrack.PublisherIdentity(),
 		"trackID", subTrack.ID(), "kind", subTrack.DownTrack().Kind())
@@ -1283,38 +1292,57 @@ func (p *ParticipantImpl) handleDataMessage(kind livekit.DataPacket_Kind, data [
 	}
 }
 
-func (p *ParticipantImpl) logSelectedCandidatePair(isPrimary bool) {
+func (p *ParticipantImpl) getTransport(isPrimary bool) *PCTransport {
 	pcTransport := p.publisher
 	if (isPrimary && p.SubscriberAsPrimary()) || (!isPrimary && !p.SubscriberAsPrimary()) {
 		pcTransport = p.subscriber
 	}
 
-	sctp := pcTransport.pc.SCTP()
-	if sctp == nil {
-		return
-	}
+	return pcTransport
+}
 
-	transport := sctp.Transport()
-	if transport == nil {
-		return
-	}
+func (p *ParticipantImpl) handleICEConnected(isPrimary bool) {
+	pcTransport := p.getTransport(isPrimary)
+	pcTransport.SetICEConnectedAt(time.Now())
 
-	iceTransport := transport.ICETransport()
-	if iceTransport == nil {
-		return
-	}
-
-	selectedCandidatePair, err := iceTransport.GetSelectedCandidatePair()
-	if err != nil {
+	if pair, err := pcTransport.GetSelectedPair(); err != nil {
 		pcTransport.Logger().Errorw("error getting selected ICE candidate pair", err)
 	} else {
-		pcTransport.Logger().Infow("selected ICE candidate pair", "pair", selectedCandidatePair)
+		pcTransport.Logger().Infow("selected ICE candidate pair", "pair", pair)
+	}
+}
+
+func (p *ParticipantImpl) handleConnectionFailed(isPrimary bool) {
+	pcTransport := p.getTransport(isPrimary)
+	isShort, duration := pcTransport.IsShortConnection(time.Now())
+	if isShort {
+		// irrespective of which one fails, force TCP on both as the other one might
+		// fail at a different time and cause another disruption
+		pair, err := pcTransport.GetSelectedPair()
+		if err != nil {
+			pcTransport.Logger().Errorw("short ICE connection", err, "duration", duration)
+		} else {
+			pcTransport.Logger().Infow("short ICE connection", "pair", pair, "duration", duration)
+		}
+		/*
+			p.lock.Lock()
+			p.iceConfig.PreferSubTcp = true
+			p.iceConfig.PreferPubTcp = true
+			p.lock.Unlock()
+		*/
+	}
+}
+
+func (p *ParticipantImpl) handleICEStateChange(isPrimary bool, state webrtc.ICEConnectionState) {
+	if state == webrtc.ICEConnectionStateConnected {
+		p.handleICEConnected(isPrimary)
+	} else if state == webrtc.ICEConnectionStateFailed {
+		p.handleConnectionFailed(isPrimary)
 	}
 }
 
 func (p *ParticipantImpl) handlePrimaryStateChange(state webrtc.PeerConnectionState) {
 	if state == webrtc.PeerConnectionStateConnected {
-		p.logSelectedCandidatePair(true)
 		if !p.firstConnected.Swap(true) {
 			p.setDowntracksConnected()
 		}
@@ -1324,6 +1352,8 @@ func (p *ParticipantImpl) handlePrimaryStateChange(state webrtc.PeerConnectionSt
 		}
 		p.incActiveCounter()
 	} else if state == webrtc.PeerConnectionStateFailed {
+		p.handleConnectionFailed(true)
+
 		// clients support resuming of connections when websocket becomes disconnected
 		p.closeSignalConnection()
 
@@ -1361,9 +1391,9 @@ func (p *ParticipantImpl) handlePrimaryStateChange(state webrtc.PeerConnectionSt
 // for the secondary peer connection, we still need to handle when they become disconnected
 // instead of allowing them to silently fail.
 func (p *ParticipantImpl) handleSecondaryStateChange(state webrtc.PeerConnectionState) {
-	if state == webrtc.PeerConnectionStateConnected {
-		p.logSelectedCandidatePair(false)
-	} else if state == webrtc.PeerConnectionStateFailed {
+	if state == webrtc.PeerConnectionStateFailed {
+		p.handleConnectionFailed(false)
+
 		// clients support resuming of connections when websocket becomes disconnected
 		p.closeSignalConnection()
 	}

pkg/rtc/transport.go

@@ -35,6 +35,8 @@ const (
 	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
+
+	shortConnectionThreshold = 2 * time.Minute
 )
 
 var (
@@ -61,6 +63,7 @@ type PCTransport struct {
 	me     *webrtc.MediaEngine
 
 	lock                  sync.RWMutex
+	iceConnectedAt        time.Time
 	pendingCandidates     []webrtc.ICECandidateInit
 	debouncedNegotiate    func(func())
 	negotiationPending    map[livekit.ParticipantID]bool
@@ -216,6 +219,43 @@ func (t *PCTransport) Logger() logger.Logger {
 	return t.params.Logger
 }
 
+func (t *PCTransport) SetICEConnectedAt(at time.Time) {
+	t.lock.Lock()
+	t.iceConnectedAt = at
+	t.lock.Unlock()
+}
+
+func (t *PCTransport) IsShortConnection(at time.Time) (bool, time.Duration) {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+
+	if t.iceConnectedAt.IsZero() {
+		return false, 0
+	}
+
+	duration := at.Sub(t.iceConnectedAt)
+	return duration < shortConnectionThreshold, duration
+}
+
+func (t *PCTransport) GetSelectedPair() (*webrtc.ICECandidatePair, error) {
+	sctp := t.pc.SCTP()
+	if sctp == nil {
+		return nil, errors.New("no SCTP")
+	}
+
+	dtlsTransport := sctp.Transport()
+	if dtlsTransport == nil {
+		return nil, errors.New("no DTLS transport")
+	}
+
+	iceTransport := dtlsTransport.ICETransport()
+	if iceTransport == nil {
+		return nil, errors.New("no ICE transport")
+	}
+
+	return iceTransport.GetSelectedCandidatePair()
+}
+
 func (t *PCTransport) createPeerConnection() error {
 	var bwe cc.BandwidthEstimator
 	pc, me, err := newPeerConnection(t.params, func(estimator cc.BandwidthEstimator) {

pkg/sfu/forwarder.go

@@ -1275,7 +1275,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in
 		if f.targetLayers.Spatial == layer {
 			if extPkt.KeyFrame || tp.switchingToTargetLayer {
 				// lock to target layer
-				f.logger.Debugw("locking to target layer", "current", f.currentLayers, "target", f.targetLayers)
+				f.logger.Infow("locking to target layer", "current", f.currentLayers, "target", f.targetLayers)
 				f.currentLayers.Spatial = f.targetLayers.Spatial
 				if !f.isTemporalSupported {
 					f.currentLayers.Temporal = f.targetLayers.Temporal