close disconnected participants when signal channel fails (#1895)

* close disconnected participants when signal channel fails

* fix typefake

* update reason

pkg/rtc/participant.go

@@ -516,6 +516,15 @@ func (p *ParticipantImpl) OnClaimsChanged(callback func(types.LocalParticipant))
 	p.lock.Unlock()
 }
 
+func (p *ParticipantImpl) HandleSignalSourceClose() {
+	p.TransportManager.SetSignalSourceValid(false)
+
+	if !p.TransportManager.HasPublisherEverConnected() && !p.TransportManager.HasSubscriberEverConnected() {
+		p.params.Logger.Infow("closing disconnected participant")
+		_ = p.Close(false, types.ParticipantCloseReasonJoinFailed, false)
+	}
+}
+
 // HandleOffer an offer from remote participant, used when clients make the initial connection
 func (p *ParticipantImpl) HandleOffer(offer webrtc.SessionDescription) {
 	p.params.Logger.Debugw("received offer", "transport", livekit.SignalTarget_PUBLISHER)

pkg/rtc/types/interfaces.go

@@ -295,6 +295,7 @@ type LocalParticipant interface {
 	CloseSignalConnection(reason SignallingCloseReason)
 	UpdateLastSeenSignal()
 	SetSignalSourceValid(valid bool)
+	HandleSignalSourceClose()
 
 	// permissions
 	ClaimGrants() *auth.ClaimGrants

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -326,6 +326,10 @@ type FakeLocalParticipant struct {
 	handleReconnectAndSendResponseReturnsOnCall map[int]struct {
 		result1 error
 	}
+	HandleSignalSourceCloseStub        func()
+	handleSignalSourceCloseMutex       sync.RWMutex
+	handleSignalSourceCloseArgsForCall []struct {
+	}
 	HasPermissionStub        func(livekit.TrackID, livekit.ParticipantIdentity) bool
 	hasPermissionMutex       sync.RWMutex
 	hasPermissionArgsForCall []struct {
@@ -2481,6 +2485,30 @@ func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseReturnsOnCall(i
 	}{result1}
 }
 
+func (fake *FakeLocalParticipant) HandleSignalSourceClose() {
+	fake.handleSignalSourceCloseMutex.Lock()
+	fake.handleSignalSourceCloseArgsForCall = append(fake.handleSignalSourceCloseArgsForCall, struct {
+	}{})
+	stub := fake.HandleSignalSourceCloseStub
+	fake.recordInvocation("HandleSignalSourceClose", []interface{}{})
+	fake.handleSignalSourceCloseMutex.Unlock()
+	if stub != nil {
+		fake.HandleSignalSourceCloseStub()
+	}
+}
+
+func (fake *FakeLocalParticipant) HandleSignalSourceCloseCallCount() int {
+	fake.handleSignalSourceCloseMutex.RLock()
+	defer fake.handleSignalSourceCloseMutex.RUnlock()
+	return len(fake.handleSignalSourceCloseArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) HandleSignalSourceCloseCalls(stub func()) {
+	fake.handleSignalSourceCloseMutex.Lock()
+	defer fake.handleSignalSourceCloseMutex.Unlock()
+	fake.HandleSignalSourceCloseStub = stub
+}
+
 func (fake *FakeLocalParticipant) HasPermission(arg1 livekit.TrackID, arg2 livekit.ParticipantIdentity) bool {
 	fake.hasPermissionMutex.Lock()
 	ret, specificReturn := fake.hasPermissionReturnsOnCall[len(fake.hasPermissionArgsForCall)]
@@ -5626,6 +5654,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.handleOfferMutex.RUnlock()
 	fake.handleReconnectAndSendResponseMutex.RLock()
 	defer fake.handleReconnectAndSendResponseMutex.RUnlock()
+	fake.handleSignalSourceCloseMutex.RLock()
+	defer fake.handleSignalSourceCloseMutex.RUnlock()
 	fake.hasPermissionMutex.RLock()
 	defer fake.hasPermissionMutex.RUnlock()
 	fake.hiddenMutex.RLock()

pkg/service/roommanager.go

@@ -549,7 +549,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.LocalPa
 			// this means ICE restart isn't possible in single node mode
 			if obj == nil {
 				if room.GetParticipantRequestSource(participant.Identity()) == requestSource {
-					participant.SetSignalSourceValid(false)
+					participant.HandleSignalSourceClose()
 				}
 				return
 			}