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

@@ -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
+}

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,
 		)
 	}

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) {

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
 

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 {

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) {

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
 		}

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()

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) {

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) {
 }