Log reason for participant close (#776)

* Log reason for participant close

Please suggest better naming for different scenarios if something comes
to mind.

* group service requests

* incorporate feedback

* Change names in tests

pkg/rtc/participant.go

@@ -648,13 +648,13 @@ func (p *ParticipantImpl) Start() {
 	})
 }
 
-func (p *ParticipantImpl) Close(sendLeave bool) error {
+func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseReason) error {
 	if p.isClosed.Swap(true) {
 		// already closed
 		return nil
 	}
 
-	p.params.Logger.Infow("closing participant", "sendLeave", sendLeave)
+	p.params.Logger.Infow("closing participant", "sendLeave", sendLeave, "reason", reason)
 	// send leave message
 	if sendLeave {
 		_ = p.writeMessage(&livekit.SignalResponse{
@@ -1207,7 +1207,7 @@ func (p *ParticipantImpl) handlePrimaryStateChange(state webrtc.PeerConnectionSt
 				}
 				if primaryPC.ConnectionState() != webrtc.PeerConnectionStateConnected {
 					p.params.Logger.Infow("closing disconnected participant")
-					p.Close(true)
+					p.Close(true, types.ParticipantCloseReasonPeerConnectionDisconnected)
 				}
 			})
 			p.lock.Unlock()

pkg/rtc/room.go

@@ -246,7 +246,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions
 			r.telemetry.ParticipantActive(context.Background(), r.ToProto(), p.ToProto(), &livekit.AnalyticsClientMeta{ClientConnectTime: uint32(time.Since(p.ConnectedAt()).Milliseconds())})
 		} else if state == livekit.ParticipantInfo_DISCONNECTED {
 			// remove participant from room
-			go r.RemoveParticipant(p.Identity())
+			go r.RemoveParticipant(p.Identity(), types.ParticipantCloseReasonStateDisconnected)
 		}
 	})
 	participant.OnTrackUpdated(r.onTrackUpdated)
@@ -302,7 +302,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions
 	time.AfterFunc(time.Minute, func() {
 		state := participant.State()
 		if state == livekit.ParticipantInfo_JOINING || state == livekit.ParticipantInfo_JOINED {
-			r.RemoveParticipant(participant.Identity())
+			r.RemoveParticipant(participant.Identity(), types.ParticipantCloseReasonJoinTimeout)
 		}
 	})
 
@@ -348,7 +348,7 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, responseSink routing.
 	return nil
 }
 
-func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity) {
+func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, reason types.ParticipantCloseReason) {
 	r.lock.Lock()
 	p, ok := r.participants[identity]
 	if ok {
@@ -391,7 +391,7 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity) {
 
 	// close participant as well
 	r.Logger.Infow("closing participant for removal", "pID", p.ID(), "participant", p.Identity())
-	_ = p.Close(true)
+	_ = p.Close(true, reason)
 
 	r.lock.RLock()
 	if len(r.participants) == 0 {
@@ -617,15 +617,20 @@ func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScen
 			Level:  0.9,
 		}})
 	case *livekit.SimulateScenario_Migration:
+		r.Logger.Infow("simulating migration", "participant", participant.Identity())
+		// drop participant without necessarily cleaning up
+		if err := participant.Close(false, types.ParticipantCloseReasonSimulateMigration); err != nil {
+			return err
+		}
 	case *livekit.SimulateScenario_NodeFailure:
 		r.Logger.Infow("simulating node failure", "participant", participant.Identity())
 		// drop participant without necessarily cleaning up
-		if err := participant.Close(false); err != nil {
+		if err := participant.Close(false, types.ParticipantCloseReasonSimulateNodeFailure); err != nil {
 			return err
 		}
 	case *livekit.SimulateScenario_ServerLeave:
 		r.Logger.Infow("simulating server leave", "participant", participant.Identity())
-		if err := participant.Close(true); err != nil {
+		if err := participant.Close(true, types.ParticipantCloseReasonSimulateServerLeave); err != nil {
 			return err
 		}
 

pkg/rtc/room_test.go

@@ -54,14 +54,14 @@ func TestJoinedState(t *testing.T) {
 		rm := newRoomWithParticipants(t, testRoomOpts{num: 1})
 		p0 := rm.GetParticipants()[0]
 		s := time.Now().Unix()
-		rm.RemoveParticipant(p0.Identity())
+		rm.RemoveParticipant(p0.Identity(), types.ParticipantCloseReasonClientRequestLeave)
 		require.Equal(t, s, rm.LastLeftAt())
 	})
 
 	t.Run("LastLeftAt should not be set when there are still participants in the room", func(t *testing.T) {
 		rm := newRoomWithParticipants(t, testRoomOpts{num: 2})
 		p0 := rm.GetParticipants()[0]
-		rm.RemoveParticipant(p0.Identity())
+		rm.RemoveParticipant(p0.Identity(), types.ParticipantCloseReasonClientRequestLeave)
 		require.EqualValues(t, 0, rm.LastLeftAt())
 	})
 }
@@ -119,7 +119,7 @@ func TestRoomJoin(t *testing.T) {
 		disconnectedParticipant := participants[1].(*typesfakes.FakeLocalParticipant)
 		disconnectedParticipant.StateReturns(livekit.ParticipantInfo_DISCONNECTED)
 
-		rm.RemoveParticipant(p.Identity())
+		rm.RemoveParticipant(p.Identity(), types.ParticipantCloseReasonStateDisconnected)
 		time.Sleep(defaultDelay)
 
 		require.Equal(t, p, changedParticipant)
@@ -332,7 +332,7 @@ func TestRoomClosure(t *testing.T) {
 		p := rm.GetParticipants()[0]
 		// allows immediate close after
 		rm.protoRoom.EmptyTimeout = 0
-		rm.RemoveParticipant(p.Identity())
+		rm.RemoveParticipant(p.Identity(), types.ParticipantCloseReasonClientRequestLeave)
 
 		time.Sleep(defaultDelay)
 

pkg/rtc/signalhandler.go

@@ -76,7 +76,7 @@ func HandleParticipantSignal(room types.Room, participant types.LocalParticipant
 		}
 	case *livekit.SignalRequest_Leave:
 		pLogger.Infow("client leaving room")
-		room.RemoveParticipant(participant.Identity())
+		room.RemoveParticipant(participant.Identity(), types.ParticipantCloseReasonClientRequestLeave)
 	case *livekit.SignalRequest_SubscriptionPermission:
 		err := room.UpdateSubscriptionPermission(participant, msg.SubscriptionPermission)
 		if err != nil {

pkg/rtc/types/interfaces.go

@@ -28,6 +28,8 @@ type AddSubscriberParams struct {
 	TrackIDs  []livekit.TrackID
 }
 
+// ---------------------------------------------
+
 type MigrateState int32
 
 const (
@@ -36,11 +38,6 @@ const (
 	MigrateStateComplete
 )
 
-type SubscribedCodecQuality struct {
-	CodecMime string
-	Quality   livekit.VideoQuality
-}
-
 func (m MigrateState) String() string {
 	switch m {
 	case MigrateStateInit:
@@ -54,6 +51,68 @@ func (m MigrateState) String() string {
 	}
 }
 
+// ---------------------------------------------
+
+type SubscribedCodecQuality struct {
+	CodecMime string
+	Quality   livekit.VideoQuality
+}
+
+// ---------------------------------------------
+
+type ParticipantCloseReason int
+
+const (
+	ParticipantCloseReasonClientRequestLeave ParticipantCloseReason = iota
+	ParticipantCloseReasonRoomManagerStop
+	ParticipantCloseReasonJoinFailed
+	ParticipantCloseReasonJoinTimeout
+	ParticipantCloseReasonRTCSessionFinish
+	ParticipantCloseReasonStateDisconnected
+	ParticipantCloseReasonPeerConnectionDisconnected
+	ParticipantCloseReasonDuplicateIdentity
+	ParticipantCloseReasonServiceRequestRemoveParticipant
+	ParticipantCloseReasonServiceRequestDeleteRoom
+	ParticipantCloseReasonSimulateMigration
+	ParticipantCloseReasonSimulateNodeFailure
+	ParticipantCloseReasonSimulateServerLeave
+)
+
+func (p ParticipantCloseReason) String() string {
+	switch p {
+	case ParticipantCloseReasonClientRequestLeave:
+		return "CLIENT_REQUEST_LEAVE"
+	case ParticipantCloseReasonRoomManagerStop:
+		return "ROOM_MANAGER_STOP"
+	case ParticipantCloseReasonJoinFailed:
+		return "JOIN_FAILED"
+	case ParticipantCloseReasonJoinTimeout:
+		return "JOIN_TIMEOUT"
+	case ParticipantCloseReasonRTCSessionFinish:
+		return "RTC_SESSION_FINISH"
+	case ParticipantCloseReasonStateDisconnected:
+		return "STATE_DISCONNECTED"
+	case ParticipantCloseReasonPeerConnectionDisconnected:
+		return "PEER_CONNECTION_DISCONNECTED"
+	case ParticipantCloseReasonDuplicateIdentity:
+		return "DUPLICATE_IDENTITY"
+	case ParticipantCloseReasonServiceRequestRemoveParticipant:
+		return "SERVICE_REQUEST_REMOVE_PARTICIPANT"
+	case ParticipantCloseReasonServiceRequestDeleteRoom:
+		return "SERVICE_REQUEST_DELETE_ROOM"
+	case ParticipantCloseReasonSimulateMigration:
+		return "SIMULATE_MIGRATION"
+	case ParticipantCloseReasonSimulateNodeFailure:
+		return "SIMULATE_NODE_FAILURE"
+	case ParticipantCloseReasonSimulateServerLeave:
+		return "SIMULATE_SERVER_LEAVE"
+	default:
+		return fmt.Sprintf("%d", int(p))
+	}
+}
+
+// ---------------------------------------------
+
 //counterfeiter:generate . Participant
 type Participant interface {
 	ID() livekit.ParticipantID
@@ -74,7 +133,7 @@ type Participant interface {
 	IsRecorder() bool
 
 	Start()
-	Close(sendLeave bool) error
+	Close(sendLeave bool, reason ParticipantCloseReason) error
 
 	SubscriptionPermission() *livekit.SubscriptionPermission
 
@@ -182,7 +241,7 @@ type LocalParticipant interface {
 type Room interface {
 	Name() livekit.RoomName
 	ID() livekit.RoomID
-	RemoveParticipant(identity livekit.ParticipantIdentity)
+	RemoveParticipant(identity livekit.ParticipantIdentity, reason ParticipantCloseReason)
 	UpdateSubscriptions(participant LocalParticipant, trackIDs []livekit.TrackID, participantTracks []*livekit.ParticipantTracks, subscribe bool) error
 	UpdateSubscriptionPermission(participant LocalParticipant, permissions *livekit.SubscriptionPermission) error
 	SyncState(participant LocalParticipant, state *livekit.SyncState) error

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -90,10 +90,11 @@ type FakeLocalParticipant struct {
 	claimGrantsReturnsOnCall map[int]struct {
 		result1 *auth.ClaimGrants
 	}
-	CloseStub        func(bool) error
+	CloseStub        func(bool, types.ParticipantCloseReason) error
 	closeMutex       sync.RWMutex
 	closeArgsForCall []struct {
 		arg1 bool
+		arg2 types.ParticipantCloseReason
 	}
 	closeReturns struct {
 		result1 error
@@ -1062,18 +1063,19 @@ func (fake *FakeLocalParticipant) ClaimGrantsReturnsOnCall(i int, result1 *auth.
 	}{result1}
 }
 
-func (fake *FakeLocalParticipant) Close(arg1 bool) error {
+func (fake *FakeLocalParticipant) Close(arg1 bool, arg2 types.ParticipantCloseReason) error {
 	fake.closeMutex.Lock()
 	ret, specificReturn := fake.closeReturnsOnCall[len(fake.closeArgsForCall)]
 	fake.closeArgsForCall = append(fake.closeArgsForCall, struct {
 		arg1 bool
-	}{arg1})
+		arg2 types.ParticipantCloseReason
+	}{arg1, arg2})
 	stub := fake.CloseStub
 	fakeReturns := fake.closeReturns
-	fake.recordInvocation("Close", []interface{}{arg1})
+	fake.recordInvocation("Close", []interface{}{arg1, arg2})
 	fake.closeMutex.Unlock()
 	if stub != nil {
-		return stub(arg1)
+		return stub(arg1, arg2)
 	}
 	if specificReturn {
 		return ret.result1
@@ -1087,17 +1089,17 @@ func (fake *FakeLocalParticipant) CloseCallCount() int {
 	return len(fake.closeArgsForCall)
 }
 
-func (fake *FakeLocalParticipant) CloseCalls(stub func(bool) error) {
+func (fake *FakeLocalParticipant) CloseCalls(stub func(bool, types.ParticipantCloseReason) error) {
 	fake.closeMutex.Lock()
 	defer fake.closeMutex.Unlock()
 	fake.CloseStub = stub
 }
 
-func (fake *FakeLocalParticipant) CloseArgsForCall(i int) bool {
+func (fake *FakeLocalParticipant) CloseArgsForCall(i int) (bool, types.ParticipantCloseReason) {
 	fake.closeMutex.RLock()
 	defer fake.closeMutex.RUnlock()
 	argsForCall := fake.closeArgsForCall[i]
-	return argsForCall.arg1
+	return argsForCall.arg1, argsForCall.arg2
 }
 
 func (fake *FakeLocalParticipant) CloseReturns(result1 error) {

pkg/rtc/types/typesfakes/fake_participant.go

@@ -23,10 +23,11 @@ type FakeParticipant struct {
 		result1 int
 		result2 error
 	}
-	CloseStub        func(bool) error
+	CloseStub        func(bool, types.ParticipantCloseReason) error
 	closeMutex       sync.RWMutex
 	closeArgsForCall []struct {
 		arg1 bool
+		arg2 types.ParticipantCloseReason
 	}
 	closeReturns struct {
 		result1 error
@@ -260,18 +261,19 @@ func (fake *FakeParticipant) AddSubscriberReturnsOnCall(i int, result1 int, resu
 	}{result1, result2}
 }
 
-func (fake *FakeParticipant) Close(arg1 bool) error {
+func (fake *FakeParticipant) Close(arg1 bool, arg2 types.ParticipantCloseReason) error {
 	fake.closeMutex.Lock()
 	ret, specificReturn := fake.closeReturnsOnCall[len(fake.closeArgsForCall)]
 	fake.closeArgsForCall = append(fake.closeArgsForCall, struct {
 		arg1 bool
-	}{arg1})
+		arg2 types.ParticipantCloseReason
+	}{arg1, arg2})
 	stub := fake.CloseStub
 	fakeReturns := fake.closeReturns
-	fake.recordInvocation("Close", []interface{}{arg1})
+	fake.recordInvocation("Close", []interface{}{arg1, arg2})
 	fake.closeMutex.Unlock()
 	if stub != nil {
-		return stub(arg1)
+		return stub(arg1, arg2)
 	}
 	if specificReturn {
 		return ret.result1
@@ -285,17 +287,17 @@ func (fake *FakeParticipant) CloseCallCount() int {
 	return len(fake.closeArgsForCall)
 }
 
-func (fake *FakeParticipant) CloseCalls(stub func(bool) error) {
+func (fake *FakeParticipant) CloseCalls(stub func(bool, types.ParticipantCloseReason) error) {
 	fake.closeMutex.Lock()
 	defer fake.closeMutex.Unlock()
 	fake.CloseStub = stub
 }
 
-func (fake *FakeParticipant) CloseArgsForCall(i int) bool {
+func (fake *FakeParticipant) CloseArgsForCall(i int) (bool, types.ParticipantCloseReason) {
 	fake.closeMutex.RLock()
 	defer fake.closeMutex.RUnlock()
 	argsForCall := fake.closeArgsForCall[i]
-	return argsForCall.arg1
+	return argsForCall.arg1, argsForCall.arg2
 }
 
 func (fake *FakeParticipant) CloseReturns(result1 error) {

pkg/rtc/types/typesfakes/fake_room.go

@@ -29,10 +29,11 @@ type FakeRoom struct {
 	nameReturnsOnCall map[int]struct {
 		result1 livekit.RoomName
 	}
-	RemoveParticipantStub        func(livekit.ParticipantIdentity)
+	RemoveParticipantStub        func(livekit.ParticipantIdentity, types.ParticipantCloseReason)
 	removeParticipantMutex       sync.RWMutex
 	removeParticipantArgsForCall []struct {
 		arg1 livekit.ParticipantIdentity
+		arg2 types.ParticipantCloseReason
 	}
 	SetParticipantPermissionStub        func(types.LocalParticipant, *livekit.ParticipantPermission) error
 	setParticipantPermissionMutex       sync.RWMutex
@@ -218,16 +219,17 @@ func (fake *FakeRoom) NameReturnsOnCall(i int, result1 livekit.RoomName) {
 	}{result1}
 }
 
-func (fake *FakeRoom) RemoveParticipant(arg1 livekit.ParticipantIdentity) {
+func (fake *FakeRoom) RemoveParticipant(arg1 livekit.ParticipantIdentity, arg2 types.ParticipantCloseReason) {
 	fake.removeParticipantMutex.Lock()
 	fake.removeParticipantArgsForCall = append(fake.removeParticipantArgsForCall, struct {
 		arg1 livekit.ParticipantIdentity
-	}{arg1})
+		arg2 types.ParticipantCloseReason
+	}{arg1, arg2})
 	stub := fake.RemoveParticipantStub
-	fake.recordInvocation("RemoveParticipant", []interface{}{arg1})
+	fake.recordInvocation("RemoveParticipant", []interface{}{arg1, arg2})
 	fake.removeParticipantMutex.Unlock()
 	if stub != nil {
-		fake.RemoveParticipantStub(arg1)
+		fake.RemoveParticipantStub(arg1, arg2)
 	}
 }
 
@@ -237,17 +239,17 @@ func (fake *FakeRoom) RemoveParticipantCallCount() int {
 	return len(fake.removeParticipantArgsForCall)
 }
 
-func (fake *FakeRoom) RemoveParticipantCalls(stub func(livekit.ParticipantIdentity)) {
+func (fake *FakeRoom) RemoveParticipantCalls(stub func(livekit.ParticipantIdentity, types.ParticipantCloseReason)) {
 	fake.removeParticipantMutex.Lock()
 	defer fake.removeParticipantMutex.Unlock()
 	fake.RemoveParticipantStub = stub
 }
 
-func (fake *FakeRoom) RemoveParticipantArgsForCall(i int) livekit.ParticipantIdentity {
+func (fake *FakeRoom) RemoveParticipantArgsForCall(i int) (livekit.ParticipantIdentity, types.ParticipantCloseReason) {
 	fake.removeParticipantMutex.RLock()
 	defer fake.removeParticipantMutex.RUnlock()
 	argsForCall := fake.removeParticipantArgsForCall[i]
-	return argsForCall.arg1
+	return argsForCall.arg1, argsForCall.arg2
 }
 
 func (fake *FakeRoom) SetParticipantPermission(arg1 types.LocalParticipant, arg2 *livekit.ParticipantPermission) error {

pkg/service/roommanager.go

@@ -166,7 +166,7 @@ func (r *RoomManager) Stop() {
 
 	for _, room := range rooms {
 		for _, p := range room.GetParticipants() {
-			_ = p.Close(true)
+			_ = p.Close(true, types.ParticipantCloseReasonRoomManagerStop)
 		}
 		room.Close()
 	}
@@ -207,8 +207,9 @@ func (r *RoomManager) StartSession(
 			)
 			return room.ResumeParticipant(participant, responseSink)
 		} else {
+			participant.GetLogger().Infow("removing duplicate participant")
 			// we need to clean up the existing participant, so a new one can join
-			room.RemoveParticipant(participant.Identity())
+			room.RemoveParticipant(participant.Identity(), types.ParticipantCloseReasonDuplicateIdentity)
 		}
 	} else if pi.Reconnect {
 		// send leave request if participant is trying to reconnect without keep subscribe state
@@ -269,7 +270,7 @@ func (r *RoomManager) StartSession(
 	}
 	if err = room.Join(participant, &opts, r.iceServersForRoom(protoRoom), r.currentNode.Region); err != nil {
 		pLogger.Errorw("could not join room", err)
-		_ = participant.Close(true)
+		_ = participant.Close(true, types.ParticipantCloseReasonJoinFailed)
 		return err
 	}
 	if err = r.roomStore.StoreParticipant(ctx, roomName, participant.ToProto()); err != nil {
@@ -389,7 +390,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.LocalPa
 			"room", room.Name(),
 			"roomID", room.ID(),
 		)
-		_ = participant.Close(true)
+		_ = participant.Close(true, types.ParticipantCloseReasonRTCSessionFinish)
 		requestSource.Close()
 	}()
 	defer rtc.Recover()
@@ -472,7 +473,7 @@ func (r *RoomManager) handleRTCMessage(ctx context.Context, roomName livekit.Roo
 			return
 		}
 		pLogger.Infow("removing participant")
-		room.RemoveParticipant(identity)
+		room.RemoveParticipant(identity, types.ParticipantCloseReasonServiceRequestRemoveParticipant)
 	case *livekit.RTCNodeMessage_MuteTrack:
 		if participant == nil {
 			return
@@ -502,7 +503,7 @@ func (r *RoomManager) handleRTCMessage(ctx context.Context, roomName livekit.Roo
 	case *livekit.RTCNodeMessage_DeleteRoom:
 		room.Logger.Infow("deleting room")
 		for _, p := range room.GetParticipants() {
-			_ = p.Close(true)
+			_ = p.Close(true, types.ParticipantCloseReasonServiceRequestDeleteRoom)
 		}
 		room.Close()
 	case *livekit.RTCNodeMessage_UpdateSubscriptions: