From 3980d049c9c3dff519de69387f26ced62aa690b5 Mon Sep 17 00:00:00 2001
From: Paul Wells <paulwe@gmail.com>
Date: Thu, 20 Jul 2023 19:23:35 -0700
Subject: [PATCH] close disconnected participants when signal channel fails
 (#1895)

* close disconnected participants when signal channel fails

* fix typefake

* update reason
---
 pkg/rtc/participant.go                        |  9 ++++++
 pkg/rtc/types/interfaces.go                   |  1 +
 .../typesfakes/fake_local_participant.go      | 30 +++++++++++++++++++
 pkg/service/roommanager.go                    |  2 +-
 4 files changed, 41 insertions(+), 1 deletion(-)

diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go
index e0d5214d4..587879833 100644
--- a/pkg/rtc/participant.go
+++ b/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)
diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go
index 443a0a3d5..eb9336121 100644
--- a/pkg/rtc/types/interfaces.go
+++ b/pkg/rtc/types/interfaces.go
@@ -295,6 +295,7 @@ type LocalParticipant interface {
 	CloseSignalConnection(reason SignallingCloseReason)
 	UpdateLastSeenSignal()
 	SetSignalSourceValid(valid bool)
+	HandleSignalSourceClose()
 
 	// permissions
 	ClaimGrants() *auth.ClaimGrants
diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go
index deb07909f..b071877d9 100644
--- a/pkg/rtc/types/typesfakes/fake_local_participant.go
+++ b/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()
diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go
index bd09448dd..b0cefd31c 100644
--- a/pkg/service/roommanager.go
+++ b/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
 			}