handle client reconnect with ICE restarts

pkg/rtc/participant.go

@@ -360,6 +360,13 @@ func (p *ParticipantImpl) Negotiate() {
 	p.subscriber.Negotiate()
 }
 
+// ICERestart restarts subscriber ICE connections
+func (p *ParticipantImpl) ICERestart() error {
+	return p.subscriber.CreateAndSendOffer(&webrtc.OfferOptions{
+		ICERestart: true,
+	})
+}
+
 // AddSubscriber subscribes op to all publishedTracks
 func (p *ParticipantImpl) AddSubscriber(op types.Participant) (int, error) {
 	p.lock.RLock()

pkg/rtc/transport.go

@@ -6,9 +6,9 @@ import (
 	"time"
 
 	"github.com/bep/debounce"
+	"github.com/livekit/livekit-server/pkg/logger"
 	"github.com/pion/webrtc/v3"
 
-	"github.com/livekit/livekit-server/pkg/logger"
 	livekit "github.com/livekit/livekit-server/proto"
 )
 
@@ -123,12 +123,16 @@ func (t *PCTransport) OnOffer(f func(sd webrtc.SessionDescription)) {
 }
 
 func (t *PCTransport) Negotiate() {
-	t.debouncedNegotiate(t.handleNegotiate)
+	t.debouncedNegotiate(func() {
+		if err := t.CreateAndSendOffer(nil); err != nil {
+			logger.Errorw("could not negotiate", "error", err)
+		}
+	})
 }
 
-func (t *PCTransport) handleNegotiate() {
+func (t *PCTransport) CreateAndSendOffer(options *webrtc.OfferOptions) error {
 	if t.onOffer == nil {
-		return
+		return nil
 	}
 
 	state := t.negotiationState.Load().(int)
@@ -136,23 +140,24 @@ func (t *PCTransport) handleNegotiate() {
 	if state == negotiationStateClient {
 		logger.Debugw("skipping negotiation, trying again later")
 		t.negotiationState.Store(negotiationRetry)
-		return
+		return nil
 	}
 
-	offer, err := t.pc.CreateOffer(nil)
+	offer, err := t.pc.CreateOffer(options)
 	if err != nil {
 		logger.Errorw("could not create offer", "err", err)
-		return
+		return err
 	}
 
 	err = t.pc.SetLocalDescription(offer)
 	if err != nil {
 		logger.Errorw("could not set local description", "err", err)
-		return
+		return err
 	}
 
 	// indicate waiting for client
 	t.negotiationState.Store(negotiationStateClient)
 
 	t.onOffer(offer)
+	return nil
 }

pkg/rtc/types/interfaces.go

@@ -36,6 +36,7 @@ type Participant interface {
 	SetResponseSink(sink routing.MessageSink)
 	SubscriberMediaEngine() *webrtc.MediaEngine
 	Negotiate()
+	ICERestart() error
 
 	AddTrack(clientId, name string, trackType livekit.TrackType)
 	GetPublishedTracks() []PublishedTrack

pkg/rtc/types/typesfakes/fake_participant.go

@@ -157,6 +157,16 @@ type FakeParticipant struct {
 		result1 webrtc.SessionDescription
 		result2 error
 	}
+	ICERestartStub        func() error
+	iCERestartMutex       sync.RWMutex
+	iCERestartArgsForCall []struct {
+	}
+	iCERestartReturns struct {
+		result1 error
+	}
+	iCERestartReturnsOnCall map[int]struct {
+		result1 error
+	}
 	IDStub        func() string
 	iDMutex       sync.RWMutex
 	iDArgsForCall []struct {
@@ -1096,6 +1106,59 @@ func (fake *FakeParticipant) HandleOfferReturnsOnCall(i int, result1 webrtc.Sess
 	}{result1, result2}
 }
 
+func (fake *FakeParticipant) ICERestart() error {
+	fake.iCERestartMutex.Lock()
+	ret, specificReturn := fake.iCERestartReturnsOnCall[len(fake.iCERestartArgsForCall)]
+	fake.iCERestartArgsForCall = append(fake.iCERestartArgsForCall, struct {
+	}{})
+	stub := fake.ICERestartStub
+	fakeReturns := fake.iCERestartReturns
+	fake.recordInvocation("ICERestart", []interface{}{})
+	fake.iCERestartMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeParticipant) ICERestartCallCount() int {
+	fake.iCERestartMutex.RLock()
+	defer fake.iCERestartMutex.RUnlock()
+	return len(fake.iCERestartArgsForCall)
+}
+
+func (fake *FakeParticipant) ICERestartCalls(stub func() error) {
+	fake.iCERestartMutex.Lock()
+	defer fake.iCERestartMutex.Unlock()
+	fake.ICERestartStub = stub
+}
+
+func (fake *FakeParticipant) ICERestartReturns(result1 error) {
+	fake.iCERestartMutex.Lock()
+	defer fake.iCERestartMutex.Unlock()
+	fake.ICERestartStub = nil
+	fake.iCERestartReturns = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *FakeParticipant) ICERestartReturnsOnCall(i int, result1 error) {
+	fake.iCERestartMutex.Lock()
+	defer fake.iCERestartMutex.Unlock()
+	fake.ICERestartStub = nil
+	if fake.iCERestartReturnsOnCall == nil {
+		fake.iCERestartReturnsOnCall = make(map[int]struct {
+			result1 error
+		})
+	}
+	fake.iCERestartReturnsOnCall[i] = struct {
+		result1 error
+	}{result1}
+}
+
 func (fake *FakeParticipant) ID() string {
 	fake.iDMutex.Lock()
 	ret, specificReturn := fake.iDReturnsOnCall[len(fake.iDArgsForCall)]
@@ -2211,6 +2274,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
 	defer fake.handleAnswerMutex.RUnlock()
 	fake.handleOfferMutex.RLock()
 	defer fake.handleOfferMutex.RUnlock()
+	fake.iCERestartMutex.RLock()
+	defer fake.iCERestartMutex.RUnlock()
 	fake.iDMutex.RLock()
 	defer fake.iDMutex.RUnlock()
 	fake.identityMutex.RLock()

pkg/service/roommanager.go

@@ -196,6 +196,12 @@ func (r *RoomManager) StartSession(roomName string, pi routing.ParticipantInit,
 				prevSink.Close()
 			}
 			participant.SetResponseSink(responseSink)
+			if err := participant.ICERestart(); err != nil {
+				logger.Warnw("could not restart ICE",
+					"participant", pi.Identity,
+					"error", err,
+				)
+			}
 			return
 		} else {
 			// we need to clean up the existing participant, so a new one can join