diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index eefa1c0e3..d0fc9a9a4 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -54,7 +54,9 @@ type MediaTrack struct { dynacastManager dynacast.DynacastManager - lock sync.RWMutex + lock sync.RWMutex + migrated bool + published bool rttFromXR atomic.Bool @@ -712,3 +714,27 @@ func (t *MediaTrack) OnDynacastSubscribedAudioCodecChange(codecs []*livekit.Subs _ = onSubscribedAudioCodecChange(t.ID(), codecs) } } + +func (t *MediaTrack) SetMigrated(migrated bool) { + t.lock.Lock() + t.migrated = migrated + t.lock.Unlock() +} + +func (t *MediaTrack) Migrated() bool { + t.lock.RLock() + defer t.lock.RUnlock() + return t.migrated +} + +func (t *MediaTrack) SetPublished(published bool) { + t.lock.Lock() + t.published = published + t.lock.Unlock() +} + +func (t *MediaTrack) Published() bool { + t.lock.RLock() + defer t.lock.RUnlock() + return t.published +} diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 2d86565f6..1cad39748 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -261,6 +261,8 @@ type ParticipantImpl struct { disconnectTimer *time.Timer migrationTimer *time.Timer + migratedInAt atomic.Pointer[time.Time] + pubRTCPQueue *sutils.TypedOpsQueue[postRtcpOp] // hold reference for MediaTrack @@ -1288,7 +1290,7 @@ func (p *ParticipantImpl) HandleAnswer(sd *livekit.SessionDescription) { func (p *ParticipantImpl) handleMigrateTracks() []*MediaTrack { // muted track won't send rtp packet, so it is required to add mediatrack manually. - // But, synthesising track publish for unmuted tracks keeps a consistent path. + // But, synthesising track publish for unmuted tracks also keeps a consistent path. // In both cases (muted and unmuted), when publisher sends media packets, OnTrack would register and go from there. var addedTracks []*MediaTrack p.pendingTracksLock.Lock() @@ -1368,6 +1370,9 @@ func (p *ParticipantImpl) SetMigrateInfo( dataChannelReceiveState []*livekit.DataChannelReceiveState, dataTracks []*livekit.PublishDataTrackResponse, ) { + now := time.Now() + p.migratedInAt.Store(pointer.To(now)) + p.pendingTracksLock.Lock() for _, t := range mediaTracks { ti := t.GetTrack() @@ -1380,7 +1385,7 @@ func (p *ParticipantImpl) SetMigrateInfo( p.pendingTracks[t.GetCid()] = &pendingTrackInfo{ trackInfos: []*livekit.TrackInfo{ti}, migrated: true, - createdAt: time.Now(), + createdAt: now, } p.pubLogger.Infow( "pending track added (migration)", @@ -1700,7 +1705,7 @@ func (p *ParticipantImpl) SetMigrateState(s types.MigrateState) { } if s == types.MigrateStateComplete { - // wait for all migrated track to be published, + // wait for all migrated tracks to be published, // it is possible that synthesized track publish above could // race with actual publish from client and the above synthesized // one could actually be a no-op because the actual publish path is active. @@ -3230,13 +3235,17 @@ func (p *ParticipantImpl) mediaTrackReceived( } // use existing media track to handle simulcast - var pubTime time.Duration + var createdAt time.Time var isMigrated bool var ridsFromSdp buffer.VideoLayersRid + var pubTime time.Duration mt, ok := p.getPublishedTrackBySdpCid(track.ID()).(*MediaTrack) if !ok { - signalCid, ti, sdpRids, migrated, createdAt := p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind()), true) - ridsFromSdp = sdpRids + var ( + signalCid string + ti *livekit.TrackInfo + ) + signalCid, ti, ridsFromSdp, isMigrated, createdAt = p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind()), true) if ti == nil { p.pendingRemoteTracks = append( p.pendingRemoteTracks, @@ -3245,10 +3254,9 @@ func (p *ParticipantImpl) mediaTrackReceived( p.pendingTracksLock.Unlock() return nil, false, false, ridsFromSdp } - isMigrated = migrated // check if the migrated track has correct codec - if migrated && len(ti.Codecs) > 0 { + if isMigrated && len(ti.Codecs) > 0 { parameters := rtpReceiver.GetParameters() var codecFound int for _, c := range ti.Codecs { @@ -3280,12 +3288,28 @@ func (p *ParticipantImpl) mediaTrackReceived( mimeType := mime.NormalizeMimeType(ti.MimeType) for _, layer := range ti.Layers { layer.SpatialLayer = buffer.VideoQualityToSpatialLayer(mimeType, layer.Quality, ti) - layer.Rid = buffer.VideoQualityToRid(mimeType, layer.Quality, ti, sdpRids) + layer.Rid = buffer.VideoQualityToRid(mimeType, layer.Quality, ti, ridsFromSdp) } mt = p.addMediaTrack(signalCid, ti) newTrack = true + } + // a track might have been set up in migrate-in path and won't show up as a new track here, + // so we need to check if it's migrated and published separately + if !isMigrated { + if isMigrated = mt.Migrated(); isMigrated { + if migratedInAt := p.migratedInAt.Load(); migratedInAt != nil { + createdAt = *migratedInAt + } + } + } + if !newTrack { + newTrack = !mt.Published() + } + mt.SetPublished(true) + + if newTrack { // if the addTrackRequest is sent before publisher peer connection is established, then it means the client tries to publish // before fully connected, in this case we only record the time when publisher peer connection is established since // we want this metric to represent the time cost by publishing. @@ -3350,6 +3374,7 @@ func (p *ParticipantImpl) addMigratedTrack(cid string, ti *livekit.TrackInfo) *M } mt := p.addMediaTrack(cid, ti) + mt.SetMigrated(true) potentialCodecs := make([]webrtc.RTPCodecParameters, 0, len(ti.Codecs)) parameters := rtpReceiver.GetParameters() @@ -3460,6 +3485,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, ti *livekit.TrackInfo) p.ID(), p.Identity(), mt.ToProto(), + mt.Published(), !isExpectedToResume, ) @@ -4129,6 +4155,7 @@ func (p *ParticipantImpl) MoveToRoom(params types.MoveToRoomParams) { p.ID(), p.Identity(), trackInfo, + track.(types.LocalMediaTrack).Published(), true, ) } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index d9d56c123..b6e97a8b0 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -125,6 +125,7 @@ type Room struct { participantOpts map[livekit.ParticipantIdentity]*ParticipantOptions participantRequestSources map[livekit.ParticipantIdentity]routing.MessageSource hasPublished map[livekit.ParticipantIdentity]bool + launchedTrackEgresses map[livekit.ParticipantIdentity][]livekit.TrackID agentParticpants map[livekit.ParticipantIdentity]*agentJob bufferFactory *buffer.FactoryOfBufferFactory @@ -273,6 +274,7 @@ func NewRoom( participantOpts: make(map[livekit.ParticipantIdentity]*ParticipantOptions), participantRequestSources: make(map[livekit.ParticipantIdentity]routing.MessageSource), hasPublished: make(map[livekit.ParticipantIdentity]bool), + launchedTrackEgresses: make(map[livekit.ParticipantIdentity][]livekit.TrackID), agentParticpants: make(map[livekit.ParticipantIdentity]*agentJob), bufferFactory: buffer.NewFactoryOfBufferFactory(config.Receiver.PacketBufferSizeVideo, config.Receiver.PacketBufferSizeAudio), batchedUpdates: make(map[livekit.ParticipantIdentity]*ParticipantUpdate), @@ -1133,19 +1135,27 @@ func (r *Room) onTrackPublished(participant types.Participant, track types.Media } } if participant.Kind() != livekit.ParticipantInfo_EGRESS && r.internal != nil && r.internal.TrackEgress != nil { - go func() { - if err := StartTrackEgress( - context.Background(), - r.egressLauncher, - r.telemetry, - r.internal.TrackEgress, - track, - r.Name(), - r.ID(), - ); err != nil { - r.logger.Errorw("failed to launch track egress", err) - } - }() + r.lock.Lock() + launchedTrackEgress := slices.Contains(r.launchedTrackEgresses[participant.Identity()], track.ID()) + if !launchedTrackEgress { + r.launchedTrackEgresses[participant.Identity()] = append(r.launchedTrackEgresses[participant.Identity()], track.ID()) + } + r.lock.Unlock() + if !launchedTrackEgress { + go func() { + if err := StartTrackEgress( + context.Background(), + r.egressLauncher, + r.telemetry, + r.internal.TrackEgress, + track, + r.Name(), + r.ID(), + ); err != nil { + r.logger.Errorw("failed to launch track egress", err) + } + }() + } } } @@ -1417,6 +1427,7 @@ func (r *Room) RemoveParticipant( delete(r.participantOpts, identity) delete(r.participantRequestSources, identity) delete(r.hasPublished, identity) + delete(r.launchedTrackEgresses, identity) delete(r.agentParticpants, identity) if !p.Hidden() { r.protoRoom.NumParticipants-- @@ -2047,8 +2058,8 @@ func (l participantTelemetryListener) OnTrackPublished(pID livekit.ParticipantID l.room.telemetry.TrackPublished(context.Background(), l.eventRoom(), pID, identity, ti, shouldSendEvent) } -func (l participantTelemetryListener) OnTrackUnpublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) { - l.room.telemetry.TrackUnpublished(context.Background(), l.eventRoom(), pID, identity, ti, shouldSendEvent) +func (l participantTelemetryListener) OnTrackUnpublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool) { + l.room.telemetry.TrackUnpublished(context.Background(), l.eventRoom(), pID, identity, ti, wasPublishedLocally, shouldSendEvent) } func (l participantTelemetryListener) OnTrackSubscribeRequested(pID livekit.ParticipantID, ti *livekit.TrackInfo) { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 92951e1ab..2b4552a51 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -682,7 +682,7 @@ func (*NullLocalParticipantListener) OnLeave(LocalParticipant, ParticipantCloseR type ParticipantTelemetryListener interface { OnTrackPublishRequested(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) OnTrackPublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) - OnTrackUnpublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) + OnTrackUnpublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool) OnTrackSubscribeRequested(pID livekit.ParticipantID, ti *livekit.TrackInfo) OnTrackSubscribed(pID livekit.ParticipantID, ti *livekit.TrackInfo, publisherInfo *livekit.ParticipantInfo, shouldSendEvent bool) OnTrackUnsubscribed(pID livekit.ParticipantID, ti *livekit.TrackInfo, shouldSendEvent bool) @@ -706,7 +706,7 @@ func (NullParticipantTelemetryListener) OnTrackPublishRequested(pID livekit.Part } func (NullParticipantTelemetryListener) OnTrackPublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) { } -func (NullParticipantTelemetryListener) OnTrackUnpublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) { +func (NullParticipantTelemetryListener) OnTrackUnpublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool) { } func (NullParticipantTelemetryListener) OnTrackSubscribeRequested(pID livekit.ParticipantID, ti *livekit.TrackInfo) { } @@ -817,6 +817,11 @@ type LocalMediaTrack interface { HasSignalCid(cid string) bool HasSdpCid(cid string) bool + SetMigrated(bool) + Migrated() bool + SetPublished(bool) + Published() bool + GetConnectionScoreAndQuality() (float32, livekit.ConnectionQuality) GetTrackStats() *livekit.RTPStats diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index d66f4299a..67b615661 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -227,6 +227,16 @@ type FakeLocalMediaTrack struct { loggerReturnsOnCall map[int]struct { result1 logger.Logger } + MigratedStub func() bool + migratedMutex sync.RWMutex + migratedArgsForCall []struct { + } + migratedReturns struct { + result1 bool + } + migratedReturnsOnCall map[int]struct { + result1 bool + } NameStub func() string nameMutex sync.RWMutex nameArgsForCall []struct { @@ -259,6 +269,16 @@ type FakeLocalMediaTrack struct { onTrackSubscribedMutex sync.RWMutex onTrackSubscribedArgsForCall []struct { } + PublishedStub func() bool + publishedMutex sync.RWMutex + publishedArgsForCall []struct { + } + publishedReturns struct { + result1 bool + } + publishedReturnsOnCall map[int]struct { + result1 bool + } PublisherIDStub func() livekit.ParticipantID publisherIDMutex sync.RWMutex publisherIDArgsForCall []struct { @@ -320,11 +340,21 @@ type FakeLocalMediaTrack struct { revokeDisallowedSubscribersReturnsOnCall map[int]struct { result1 []livekit.ParticipantIdentity } + SetMigratedStub func(bool) + setMigratedMutex sync.RWMutex + setMigratedArgsForCall []struct { + arg1 bool + } SetMutedStub func(bool) setMutedMutex sync.RWMutex setMutedArgsForCall []struct { arg1 bool } + SetPublishedStub func(bool) + setPublishedMutex sync.RWMutex + setPublishedArgsForCall []struct { + arg1 bool + } SetRTTStub func(uint32) setRTTMutex sync.RWMutex setRTTArgsForCall []struct { @@ -1514,6 +1544,59 @@ func (fake *FakeLocalMediaTrack) LoggerReturnsOnCall(i int, result1 logger.Logge }{result1} } +func (fake *FakeLocalMediaTrack) Migrated() bool { + fake.migratedMutex.Lock() + ret, specificReturn := fake.migratedReturnsOnCall[len(fake.migratedArgsForCall)] + fake.migratedArgsForCall = append(fake.migratedArgsForCall, struct { + }{}) + stub := fake.MigratedStub + fakeReturns := fake.migratedReturns + fake.recordInvocation("Migrated", []interface{}{}) + fake.migratedMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalMediaTrack) MigratedCallCount() int { + fake.migratedMutex.RLock() + defer fake.migratedMutex.RUnlock() + return len(fake.migratedArgsForCall) +} + +func (fake *FakeLocalMediaTrack) MigratedCalls(stub func() bool) { + fake.migratedMutex.Lock() + defer fake.migratedMutex.Unlock() + fake.MigratedStub = stub +} + +func (fake *FakeLocalMediaTrack) MigratedReturns(result1 bool) { + fake.migratedMutex.Lock() + defer fake.migratedMutex.Unlock() + fake.MigratedStub = nil + fake.migratedReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalMediaTrack) MigratedReturnsOnCall(i int, result1 bool) { + fake.migratedMutex.Lock() + defer fake.migratedMutex.Unlock() + fake.MigratedStub = nil + if fake.migratedReturnsOnCall == nil { + fake.migratedReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.migratedReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalMediaTrack) Name() string { fake.nameMutex.Lock() ret, specificReturn := fake.nameReturnsOnCall[len(fake.nameArgsForCall)] @@ -1700,6 +1783,59 @@ func (fake *FakeLocalMediaTrack) OnTrackSubscribedCalls(stub func()) { fake.OnTrackSubscribedStub = stub } +func (fake *FakeLocalMediaTrack) Published() bool { + fake.publishedMutex.Lock() + ret, specificReturn := fake.publishedReturnsOnCall[len(fake.publishedArgsForCall)] + fake.publishedArgsForCall = append(fake.publishedArgsForCall, struct { + }{}) + stub := fake.PublishedStub + fakeReturns := fake.publishedReturns + fake.recordInvocation("Published", []interface{}{}) + fake.publishedMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalMediaTrack) PublishedCallCount() int { + fake.publishedMutex.RLock() + defer fake.publishedMutex.RUnlock() + return len(fake.publishedArgsForCall) +} + +func (fake *FakeLocalMediaTrack) PublishedCalls(stub func() bool) { + fake.publishedMutex.Lock() + defer fake.publishedMutex.Unlock() + fake.PublishedStub = stub +} + +func (fake *FakeLocalMediaTrack) PublishedReturns(result1 bool) { + fake.publishedMutex.Lock() + defer fake.publishedMutex.Unlock() + fake.PublishedStub = nil + fake.publishedReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalMediaTrack) PublishedReturnsOnCall(i int, result1 bool) { + fake.publishedMutex.Lock() + defer fake.publishedMutex.Unlock() + fake.PublishedStub = nil + if fake.publishedReturnsOnCall == nil { + fake.publishedReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.publishedReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalMediaTrack) PublisherID() livekit.ParticipantID { fake.publisherIDMutex.Lock() ret, specificReturn := fake.publisherIDReturnsOnCall[len(fake.publisherIDArgsForCall)] @@ -2035,6 +2171,38 @@ func (fake *FakeLocalMediaTrack) RevokeDisallowedSubscribersReturnsOnCall(i int, }{result1} } +func (fake *FakeLocalMediaTrack) SetMigrated(arg1 bool) { + fake.setMigratedMutex.Lock() + fake.setMigratedArgsForCall = append(fake.setMigratedArgsForCall, struct { + arg1 bool + }{arg1}) + stub := fake.SetMigratedStub + fake.recordInvocation("SetMigrated", []interface{}{arg1}) + fake.setMigratedMutex.Unlock() + if stub != nil { + fake.SetMigratedStub(arg1) + } +} + +func (fake *FakeLocalMediaTrack) SetMigratedCallCount() int { + fake.setMigratedMutex.RLock() + defer fake.setMigratedMutex.RUnlock() + return len(fake.setMigratedArgsForCall) +} + +func (fake *FakeLocalMediaTrack) SetMigratedCalls(stub func(bool)) { + fake.setMigratedMutex.Lock() + defer fake.setMigratedMutex.Unlock() + fake.SetMigratedStub = stub +} + +func (fake *FakeLocalMediaTrack) SetMigratedArgsForCall(i int) bool { + fake.setMigratedMutex.RLock() + defer fake.setMigratedMutex.RUnlock() + argsForCall := fake.setMigratedArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalMediaTrack) SetMuted(arg1 bool) { fake.setMutedMutex.Lock() fake.setMutedArgsForCall = append(fake.setMutedArgsForCall, struct { @@ -2067,6 +2235,38 @@ func (fake *FakeLocalMediaTrack) SetMutedArgsForCall(i int) bool { return argsForCall.arg1 } +func (fake *FakeLocalMediaTrack) SetPublished(arg1 bool) { + fake.setPublishedMutex.Lock() + fake.setPublishedArgsForCall = append(fake.setPublishedArgsForCall, struct { + arg1 bool + }{arg1}) + stub := fake.SetPublishedStub + fake.recordInvocation("SetPublished", []interface{}{arg1}) + fake.setPublishedMutex.Unlock() + if stub != nil { + fake.SetPublishedStub(arg1) + } +} + +func (fake *FakeLocalMediaTrack) SetPublishedCallCount() int { + fake.setPublishedMutex.RLock() + defer fake.setPublishedMutex.RUnlock() + return len(fake.setPublishedArgsForCall) +} + +func (fake *FakeLocalMediaTrack) SetPublishedCalls(stub func(bool)) { + fake.setPublishedMutex.Lock() + defer fake.setPublishedMutex.Unlock() + fake.SetPublishedStub = stub +} + +func (fake *FakeLocalMediaTrack) SetPublishedArgsForCall(i int) bool { + fake.setPublishedMutex.RLock() + defer fake.setPublishedMutex.RUnlock() + argsForCall := fake.setPublishedArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalMediaTrack) SetRTT(arg1 uint32) { fake.setRTTMutex.Lock() fake.setRTTArgsForCall = append(fake.setRTTArgsForCall, struct { diff --git a/pkg/rtc/types/typesfakes/fake_participant_telemetry_listener.go b/pkg/rtc/types/typesfakes/fake_participant_telemetry_listener.go index 7b7bff499..68aa8c21a 100644 --- a/pkg/rtc/types/typesfakes/fake_participant_telemetry_listener.go +++ b/pkg/rtc/types/typesfakes/fake_participant_telemetry_listener.go @@ -106,13 +106,14 @@ type FakeParticipantTelemetryListener struct { arg1 livekit.ParticipantID arg2 *livekit.TrackInfo } - OnTrackUnpublishedStub func(livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) + OnTrackUnpublishedStub func(livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool) onTrackUnpublishedMutex sync.RWMutex onTrackUnpublishedArgsForCall []struct { arg1 livekit.ParticipantID arg2 livekit.ParticipantIdentity arg3 *livekit.TrackInfo arg4 bool + arg5 bool } OnTrackUnsubscribedStub func(livekit.ParticipantID, *livekit.TrackInfo, bool) onTrackUnsubscribedMutex sync.RWMutex @@ -570,19 +571,20 @@ func (fake *FakeParticipantTelemetryListener) OnTrackUnmutedArgsForCall(i int) ( return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakeParticipantTelemetryListener) OnTrackUnpublished(arg1 livekit.ParticipantID, arg2 livekit.ParticipantIdentity, arg3 *livekit.TrackInfo, arg4 bool) { +func (fake *FakeParticipantTelemetryListener) OnTrackUnpublished(arg1 livekit.ParticipantID, arg2 livekit.ParticipantIdentity, arg3 *livekit.TrackInfo, arg4 bool, arg5 bool) { fake.onTrackUnpublishedMutex.Lock() fake.onTrackUnpublishedArgsForCall = append(fake.onTrackUnpublishedArgsForCall, struct { arg1 livekit.ParticipantID arg2 livekit.ParticipantIdentity arg3 *livekit.TrackInfo arg4 bool - }{arg1, arg2, arg3, arg4}) + arg5 bool + }{arg1, arg2, arg3, arg4, arg5}) stub := fake.OnTrackUnpublishedStub - fake.recordInvocation("OnTrackUnpublished", []interface{}{arg1, arg2, arg3, arg4}) + fake.recordInvocation("OnTrackUnpublished", []interface{}{arg1, arg2, arg3, arg4, arg5}) fake.onTrackUnpublishedMutex.Unlock() if stub != nil { - fake.OnTrackUnpublishedStub(arg1, arg2, arg3, arg4) + fake.OnTrackUnpublishedStub(arg1, arg2, arg3, arg4, arg5) } } @@ -592,17 +594,17 @@ func (fake *FakeParticipantTelemetryListener) OnTrackUnpublishedCallCount() int return len(fake.onTrackUnpublishedArgsForCall) } -func (fake *FakeParticipantTelemetryListener) OnTrackUnpublishedCalls(stub func(livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool)) { +func (fake *FakeParticipantTelemetryListener) OnTrackUnpublishedCalls(stub func(livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool)) { fake.onTrackUnpublishedMutex.Lock() defer fake.onTrackUnpublishedMutex.Unlock() fake.OnTrackUnpublishedStub = stub } -func (fake *FakeParticipantTelemetryListener) OnTrackUnpublishedArgsForCall(i int) (livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) { +func (fake *FakeParticipantTelemetryListener) OnTrackUnpublishedArgsForCall(i int) (livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool) { fake.onTrackUnpublishedMutex.RLock() defer fake.onTrackUnpublishedMutex.RUnlock() argsForCall := fake.onTrackUnpublishedArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5 } func (fake *FakeParticipantTelemetryListener) OnTrackUnsubscribed(arg1 livekit.ParticipantID, arg2 *livekit.TrackInfo, arg3 bool) { diff --git a/pkg/telemetry/events.go b/pkg/telemetry/events.go index 862d8522e..bf684156e 100644 --- a/pkg/telemetry/events.go +++ b/pkg/telemetry/events.go @@ -380,10 +380,13 @@ func (t *telemetryService) TrackUnpublished( participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, + wasPublishedLocally bool, shouldSendEvent bool, ) { t.enqueue(func() { - prometheus.SubPublishedTrack(track.Type.String()) + if wasPublishedLocally { + prometheus.SubPublishedTrack(track.Type.String()) + } if !shouldSendEvent { return } diff --git a/pkg/telemetry/stats_test.go b/pkg/telemetry/stats_test.go index 26f7835ae..4729d99dd 100644 --- a/pkg/telemetry/stats_test.go +++ b/pkg/telemetry/stats_test.go @@ -479,7 +479,7 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) { require.True(t, found2) // remove 1 track - track stats were flushed above, so no more calls to SendStats - fixture.sut.TrackUnpublished(context.Background(), room, partSID, identity, &livekit.TrackInfo{Sid: string(trackID2)}, true) + fixture.sut.TrackUnpublished(context.Background(), room, partSID, identity, &livekit.TrackInfo{Sid: string(trackID2)}, true, true) // flush fixture.flush() diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index b2363873c..5c0f28881 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go @@ -280,7 +280,7 @@ type FakeTelemetryService struct { arg3 livekit.ParticipantID arg4 *livekit.TrackInfo } - TrackUnpublishedStub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) + TrackUnpublishedStub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool) trackUnpublishedMutex sync.RWMutex trackUnpublishedArgsForCall []struct { arg1 context.Context @@ -289,6 +289,7 @@ type FakeTelemetryService struct { arg4 livekit.ParticipantIdentity arg5 *livekit.TrackInfo arg6 bool + arg7 bool } TrackUnsubscribedStub func(context.Context, *livekit.Room, livekit.ParticipantID, *livekit.TrackInfo, bool) trackUnsubscribedMutex sync.RWMutex @@ -1543,7 +1544,7 @@ func (fake *FakeTelemetryService) TrackUnmutedArgsForCall(i int) (context.Contex return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 } -func (fake *FakeTelemetryService) TrackUnpublished(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.ParticipantID, arg4 livekit.ParticipantIdentity, arg5 *livekit.TrackInfo, arg6 bool) { +func (fake *FakeTelemetryService) TrackUnpublished(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.ParticipantID, arg4 livekit.ParticipantIdentity, arg5 *livekit.TrackInfo, arg6 bool, arg7 bool) { fake.trackUnpublishedMutex.Lock() fake.trackUnpublishedArgsForCall = append(fake.trackUnpublishedArgsForCall, struct { arg1 context.Context @@ -1552,12 +1553,13 @@ func (fake *FakeTelemetryService) TrackUnpublished(arg1 context.Context, arg2 *l arg4 livekit.ParticipantIdentity arg5 *livekit.TrackInfo arg6 bool - }{arg1, arg2, arg3, arg4, arg5, arg6}) + arg7 bool + }{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) stub := fake.TrackUnpublishedStub - fake.recordInvocation("TrackUnpublished", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6}) + fake.recordInvocation("TrackUnpublished", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) fake.trackUnpublishedMutex.Unlock() if stub != nil { - fake.TrackUnpublishedStub(arg1, arg2, arg3, arg4, arg5, arg6) + fake.TrackUnpublishedStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7) } } @@ -1567,17 +1569,17 @@ func (fake *FakeTelemetryService) TrackUnpublishedCallCount() int { return len(fake.trackUnpublishedArgsForCall) } -func (fake *FakeTelemetryService) TrackUnpublishedCalls(stub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool)) { +func (fake *FakeTelemetryService) TrackUnpublishedCalls(stub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool)) { fake.trackUnpublishedMutex.Lock() defer fake.trackUnpublishedMutex.Unlock() fake.TrackUnpublishedStub = stub } -func (fake *FakeTelemetryService) TrackUnpublishedArgsForCall(i int) (context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) { +func (fake *FakeTelemetryService) TrackUnpublishedArgsForCall(i int) (context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool) { fake.trackUnpublishedMutex.RLock() defer fake.trackUnpublishedMutex.RUnlock() argsForCall := fake.trackUnpublishedArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7 } func (fake *FakeTelemetryService) TrackUnsubscribed(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.ParticipantID, arg4 *livekit.TrackInfo, arg5 bool) { diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 134e85d35..28c90400b 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -50,7 +50,7 @@ type TelemetryService interface { // TrackPublished - a publication attempt has been successful TrackPublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) // TrackUnpublished - a participant unpublished a track - TrackUnpublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) + TrackUnpublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool) // TrackSubscribeRequested - a participant requested to subscribe to a track TrackSubscribeRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, track *livekit.TrackInfo) // TrackSubscribed - a participant subscribed to a track successfully @@ -118,7 +118,7 @@ func (n NullTelemetryService) TrackPublishRequested(ctx context.Context, room *l } func (n NullTelemetryService) TrackPublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) { } -func (n NullTelemetryService) TrackUnpublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) { +func (n NullTelemetryService) TrackUnpublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool) { } func (n NullTelemetryService) TrackSubscribeRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, track *livekit.TrackInfo) { }