diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 604631373..fe117a37b 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -39,6 +39,7 @@ import ( "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" "github.com/livekit/protocol/utils/guid" + "github.com/livekit/protocol/utils/pointer" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/metric" @@ -239,7 +240,7 @@ type ParticipantImpl struct { onTrackPublished func(types.LocalParticipant, types.MediaTrack) onTrackUpdated func(types.LocalParticipant, types.MediaTrack) onTrackUnpublished func(types.LocalParticipant, types.MediaTrack) - onStateChange func(p types.LocalParticipant, state livekit.ParticipantInfo_State) + onStateChange func(p types.LocalParticipant) onMigrateStateChange func(p types.LocalParticipant, migrateState types.MigrateState) onParticipantUpdate func(types.LocalParticipant) onDataPacket func(types.LocalParticipant, livekit.DataPacket_Kind, *livekit.DataPacket) @@ -739,13 +740,13 @@ func (p *ParticipantImpl) getOnTrackUnpublished() func(types.LocalParticipant, t return p.onTrackUnpublished } -func (p *ParticipantImpl) OnStateChange(callback func(p types.LocalParticipant, state livekit.ParticipantInfo_State)) { +func (p *ParticipantImpl) OnStateChange(callback func(p types.LocalParticipant)) { p.lock.Lock() p.onStateChange = callback p.lock.Unlock() } -func (p *ParticipantImpl) getOnStateChange() func(p types.LocalParticipant, state livekit.ParticipantInfo_State) { +func (p *ParticipantImpl) getOnStateChange() func(p types.LocalParticipant) { p.lock.RLock() defer p.lock.RUnlock() return p.onStateChange @@ -1738,23 +1739,30 @@ func (p *ParticipantImpl) setupMetrics() { } func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) { - if state == livekit.ParticipantInfo_ACTIVE { - t := time.Now() - p.lastActiveAt.CompareAndSwap(nil, &t) - } - oldState := p.state.Swap(state).(livekit.ParticipantInfo_State) - if oldState == state { - return + var oldState livekit.ParticipantInfo_State + for { + oldState = p.state.Load().(livekit.ParticipantInfo_State) + if state <= oldState { + p.params.Logger.Debugw("ignoring out of order participant state", "state", state.String()) + return + } + if state == livekit.ParticipantInfo_ACTIVE { + p.lastActiveAt.CompareAndSwap(nil, pointer.To(time.Now())) + } + if p.state.CompareAndSwap(oldState, state) { + break + } } - if state == livekit.ParticipantInfo_DISCONNECTED && oldState == livekit.ParticipantInfo_ACTIVE { - prometheus.RecordSessionDuration(int(p.ProtocolVersion()), time.Since(*p.lastActiveAt.Load())) - } p.params.Logger.Debugw("updating participant state", "state", state.String()) p.dirty.Store(true) if onStateChange := p.getOnStateChange(); onStateChange != nil { - go onStateChange(p, state) + go onStateChange(p) + } + + if state == livekit.ParticipantInfo_DISCONNECTED && oldState == livekit.ParticipantInfo_ACTIVE { + prometheus.RecordSessionDuration(int(p.ProtocolVersion()), time.Since(*p.lastActiveAt.Load())) } } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index eda9519df..5b654bc5a 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -444,13 +444,16 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me r.joinedAt.Store(time.Now().Unix()) } - participant.OnStateChange(func(p types.LocalParticipant, state livekit.ParticipantInfo_State) { + var onStateChangeMu sync.Mutex + participant.OnStateChange(func(p types.LocalParticipant) { if r.onParticipantChanged != nil { r.onParticipantChanged(p) } r.broadcastParticipantState(p, broadcastOptions{skipSource: true}) - if state == livekit.ParticipantInfo_ACTIVE { + onStateChangeMu.Lock() + defer onStateChangeMu.Unlock() + if state := p.State(); state == livekit.ParticipantInfo_ACTIVE { // subscribe participant to existing published tracks r.subscribeToExistingTracks(p) diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index f502e9ca5..83bcc563d 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -111,7 +111,8 @@ func TestRoomJoin(t *testing.T) { stateChangeCB := p.OnStateChangeArgsForCall(0) require.NotNil(t, stateChangeCB) - stateChangeCB(p, livekit.ParticipantInfo_ACTIVE) + p.StateReturns(livekit.ParticipantInfo_ACTIVE) + stateChangeCB(p) // it should become a subscriber when connectivity changes numTracks := 0 @@ -747,7 +748,8 @@ func TestHiddenParticipants(t *testing.T) { stateChangeCB := hidden.OnStateChangeArgsForCall(0) require.NotNil(t, stateChangeCB) - stateChangeCB(hidden, livekit.ParticipantInfo_ACTIVE) + hidden.StateReturns(livekit.ParticipantInfo_ACTIVE) + stateChangeCB(hidden) require.Eventually(t, func() bool { return hidden.SubscribeToTrackCallCount() == 2 }, 5*time.Second, 10*time.Millisecond) }) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index c840681db..7dca0d0c2 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -414,7 +414,7 @@ type LocalParticipant interface { IssueFullReconnect(reason ParticipantCloseReason) // callbacks - OnStateChange(func(p LocalParticipant, state livekit.ParticipantInfo_State)) + OnStateChange(func(p LocalParticipant)) OnMigrateStateChange(func(p LocalParticipant, migrateState MigrateState)) // OnTrackPublished - remote added a track OnTrackPublished(func(LocalParticipant, MediaTrack)) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index e0fbfa826..55c675f00 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -726,10 +726,10 @@ type FakeLocalParticipant struct { onParticipantUpdateArgsForCall []struct { arg1 func(types.LocalParticipant) } - OnStateChangeStub func(func(p types.LocalParticipant, state livekit.ParticipantInfo_State)) + OnStateChangeStub func(func(p types.LocalParticipant)) onStateChangeMutex sync.RWMutex onStateChangeArgsForCall []struct { - arg1 func(p types.LocalParticipant, state livekit.ParticipantInfo_State) + arg1 func(p types.LocalParticipant) } OnSubscribeStatusChangedStub func(func(publisherID livekit.ParticipantID, subscribed bool)) onSubscribeStatusChangedMutex sync.RWMutex @@ -4975,10 +4975,10 @@ func (fake *FakeLocalParticipant) OnParticipantUpdateArgsForCall(i int) func(typ return argsForCall.arg1 } -func (fake *FakeLocalParticipant) OnStateChange(arg1 func(p types.LocalParticipant, state livekit.ParticipantInfo_State)) { +func (fake *FakeLocalParticipant) OnStateChange(arg1 func(p types.LocalParticipant)) { fake.onStateChangeMutex.Lock() fake.onStateChangeArgsForCall = append(fake.onStateChangeArgsForCall, struct { - arg1 func(p types.LocalParticipant, state livekit.ParticipantInfo_State) + arg1 func(p types.LocalParticipant) }{arg1}) stub := fake.OnStateChangeStub fake.recordInvocation("OnStateChange", []interface{}{arg1}) @@ -4994,13 +4994,13 @@ func (fake *FakeLocalParticipant) OnStateChangeCallCount() int { return len(fake.onStateChangeArgsForCall) } -func (fake *FakeLocalParticipant) OnStateChangeCalls(stub func(func(p types.LocalParticipant, state livekit.ParticipantInfo_State))) { +func (fake *FakeLocalParticipant) OnStateChangeCalls(stub func(func(p types.LocalParticipant))) { fake.onStateChangeMutex.Lock() defer fake.onStateChangeMutex.Unlock() fake.OnStateChangeStub = stub } -func (fake *FakeLocalParticipant) OnStateChangeArgsForCall(i int) func(p types.LocalParticipant, state livekit.ParticipantInfo_State) { +func (fake *FakeLocalParticipant) OnStateChangeArgsForCall(i int) func(p types.LocalParticipant) { fake.onStateChangeMutex.RLock() defer fake.onStateChangeMutex.RUnlock() argsForCall := fake.onStateChangeArgsForCall[i]