From d279899b7c10bf0e17c39cb63c95175ec0ecaeb4 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 11 Aug 2026 16:13:57 +0530 Subject: [PATCH] Fix publish track count on migration in. (#4740) * Fix publish track count on migration in. https://github.com/livekit/livekit/pull/4707 addressed the case of publish tracks overcounting due to synthesised track publish on migrate in. But, it introduced an issue where published tracks count could go negative because unpublish subtracted the counter irrespective of the track actually migrated in or not. Fix it by keeping track of local publish. Also, the older code was skipping publisher track count increase if the synthesised publish was handled first. Address it by checking if the track is actually new (i. e. fresh local publish) when the track was already created in the migrate in path. * fix pub time for tracks published after migration * test * prevent multiple track egresses --- pkg/rtc/mediatrack.go | 28 ++- pkg/rtc/participant.go | 45 +++- pkg/rtc/room.go | 41 ++-- pkg/rtc/types/interfaces.go | 9 +- .../typesfakes/fake_local_media_track.go | 200 ++++++++++++++++++ .../fake_participant_telemetry_listener.go | 18 +- pkg/telemetry/events.go | 5 +- pkg/telemetry/stats_test.go | 2 +- .../telemetryfakes/fake_telemetry_service.go | 18 +- pkg/telemetry/telemetryservice.go | 4 +- 10 files changed, 323 insertions(+), 47 deletions(-) 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) { }