Move subscribe/unsubscribe queue to participant. (#813)

* Move subscribe/unsubscribe queue to participant.

As subscribe/unsubscribe operation can come from both
local media track or remote media track, participant
needs to have it.

* Remove comment

* Stop reneg timer on close

* address comments

pkg/rtc/mediatrack.go

@@ -38,6 +38,7 @@ type MediaTrackParams struct {
 	SdpCid              string
 	ParticipantID       livekit.ParticipantID
 	ParticipantIdentity livekit.ParticipantIdentity
+	ParticipantVersion  uint32
 	// channel to send RTCP packets to the source
 	RTCPChan          chan []rtcp.Packet
 	BufferFactory     *buffer.Factory
@@ -61,6 +62,7 @@ func NewMediaTrack(params MediaTrackParams) *MediaTrack {
 		MediaTrack:          t,
 		ParticipantID:       params.ParticipantID,
 		ParticipantIdentity: params.ParticipantIdentity,
+		ParticipantVersion:  params.ParticipantVersion,
 		BufferFactory:       params.BufferFactory,
 		ReceiverConfig:      params.ReceiverConfig,
 		SubscriberConfig:    params.SubscriberConfig,

pkg/rtc/mediatrackreceiver.go

@@ -38,6 +38,20 @@ func (r *simulcastReceiver) Priority() int {
 	return r.priority
 }
 
+type MediaTrackReceiverParams struct {
+	TrackInfo           *livekit.TrackInfo
+	MediaTrack          types.MediaTrack
+	ParticipantID       livekit.ParticipantID
+	ParticipantIdentity livekit.ParticipantIdentity
+	ParticipantVersion  uint32
+	BufferFactory       *buffer.Factory
+	ReceiverConfig      ReceiverConfig
+	SubscriberConfig    DirectionConfig
+	VideoConfig         config.VideoConfig
+	Telemetry           telemetry.TelemetryService
+	Logger              logger.Logger
+}
+
 type MediaTrackReceiver struct {
 	params      MediaTrackReceiverParams
 	muted       atomic.Bool
@@ -62,19 +76,6 @@ type MediaTrackReceiver struct {
 	*MediaTrackSubscriptions
 }
 
-type MediaTrackReceiverParams struct {
-	TrackInfo           *livekit.TrackInfo
-	MediaTrack          types.MediaTrack
-	ParticipantID       livekit.ParticipantID
-	ParticipantIdentity livekit.ParticipantIdentity
-	BufferFactory       *buffer.Factory
-	ReceiverConfig      ReceiverConfig
-	SubscriberConfig    DirectionConfig
-	VideoConfig         config.VideoConfig
-	Telemetry           telemetry.TelemetryService
-	Logger              logger.Logger
-}
-
 func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver {
 	t := &MediaTrackReceiver{
 		params:          params,
@@ -92,6 +93,9 @@ func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver
 		Logger:           params.Logger,
 	})
 	t.MediaTrackSubscriptions.OnDownTrackCreated(t.onDownTrackCreated)
+	t.MediaTrackSubscriptions.OnSubscriptionOperationComplete(func(sub types.LocalParticipant) {
+		go sub.ClearInProgressAndProcessSubscriptionRequestsQueue(t.ID())
+	})
 
 	if t.trackInfo.Muted {
 		t.SetMuted(true)
@@ -305,6 +309,10 @@ func (t *MediaTrackReceiver) PublisherIdentity() livekit.ParticipantIdentity {
 	return t.params.ParticipantIdentity
 }
 
+func (t *MediaTrackReceiver) PublisherVersion() uint32 {
+	return t.params.ParticipantVersion
+}
+
 func (t *MediaTrackReceiver) IsSimulcast() bool {
 	return t.simulcasted.Load()
 }
@@ -349,6 +357,12 @@ func (t *MediaTrackReceiver) AddOnClose(f func()) {
 
 // AddSubscriber subscribes sub to current mediaTrack
 func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) error {
+	trackID := t.ID()
+	sub.EnqueueSubscribeTrack(trackID, t.addSubscriber)
+	return nil
+}
+
+func (t *MediaTrackReceiver) addSubscriber(sub types.LocalParticipant) error {
 	t.lock.RLock()
 	receivers := t.receiversShadow
 	potentialCodecs := make([]webrtc.RTPCodecParameters, len(t.potentialCodecs))
@@ -390,6 +404,52 @@ func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) error {
 	return nil
 }
 
+// RemoveSubscriber removes participant from subscription
+// stop all forwarders to the client
+func (t *MediaTrackReceiver) RemoveSubscriber(subscriberID livekit.ParticipantID, willBeResumed bool) {
+	subTrack := t.getSubscribedTrack(subscriberID)
+	if subTrack == nil {
+		return
+	}
+
+	sub := subTrack.Subscriber()
+	trackID := subTrack.ID()
+	sub.EnqueueUnsubscribeTrack(trackID, willBeResumed, t.MediaTrackSubscriptions.RemoveSubscriber)
+}
+
+func (t *MediaTrackReceiver) RemoveAllSubscribers(willBeResumed bool) {
+	t.params.Logger.Debugw("removing all subscribers")
+	for _, subscriberID := range t.MediaTrackSubscriptions.GetAllSubscribers() {
+		t.RemoveSubscriber(subscriberID, willBeResumed)
+	}
+}
+
+func (t *MediaTrackReceiver) RevokeDisallowedSubscribers(allowedSubscriberIdentities []livekit.ParticipantIdentity) []livekit.ParticipantIdentity {
+	var revokedSubscriberIdentities []livekit.ParticipantIdentity
+
+	// LK-TODO: large number of subscribers needs to be solved for this loop
+	for _, subTrack := range t.MediaTrackSubscriptions.getAllSubscribedTracks() {
+		found := false
+		for _, allowedIdentity := range allowedSubscriberIdentities {
+			if subTrack.SubscriberIdentity() == allowedIdentity {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			t.params.Logger.Infow("revoking subscription",
+				"subscriber", subTrack.SubscriberIdentity(),
+				"subscriberID", subTrack.SubscriberID(),
+			)
+			t.RemoveSubscriber(subTrack.SubscriberID(), false)
+			revokedSubscriberIdentities = append(revokedSubscriberIdentities, subTrack.SubscriberIdentity())
+		}
+	}
+
+	return revokedSubscriberIdentities
+}
+
 func (t *MediaTrackReceiver) UpdateTrackInfo(ti *livekit.TrackInfo) {
 	t.lock.Lock()
 	t.trackInfo = proto.Clone(ti).(*livekit.TrackInfo)

pkg/rtc/mediatracksubscriptions.go

@@ -34,30 +34,12 @@ var (
 	errNotFound          = errors.New("not found")
 )
 
-type SubscribeRequestType int
-
-const (
-	SubscribeRequestTypeRemove SubscribeRequestType = iota
-	SubscribeRequestTypeAdd
-)
-
-type SubscribeRequest struct {
-	requestType   SubscribeRequestType
-	sub           types.LocalParticipant
-	wr            *WrappedReceiver
-	willBeResumed bool
-}
-
 // MediaTrackSubscriptions manages subscriptions of a media track
 type MediaTrackSubscriptions struct {
 	params MediaTrackSubscriptionsParams
 
 	subscribedTracksMu sync.RWMutex
 	subscribedTracks   map[livekit.ParticipantID]types.SubscribedTrack
-	inProgress         map[livekit.ParticipantID]bool
-	requestsQueue      map[livekit.ParticipantID][]SubscribeRequest
-
-	onNoSubscribers func()
 
 	// quality level enable/disable
 	maxQualityLock               sync.RWMutex
@@ -70,7 +52,8 @@ type MediaTrackSubscriptions struct {
 
 	qualityNotifyOpQueue *utils.OpsQueue
 
-	onDownTrackCreated func(downTrack *sfu.DownTrack)
+	onDownTrackCreated              func(downTrack *sfu.DownTrack)
+	onSubscriptionOperationComplete func(sub types.LocalParticipant)
 }
 
 type MediaTrackSubscriptionsParams struct {
@@ -90,8 +73,6 @@ func NewMediaTrackSubscriptions(params MediaTrackSubscriptionsParams) *MediaTrac
 	t := &MediaTrackSubscriptions{
 		params:                       params,
 		subscribedTracks:             make(map[livekit.ParticipantID]types.SubscribedTrack),
-		inProgress:                   make(map[livekit.ParticipantID]bool),
-		requestsQueue:                make(map[livekit.ParticipantID][]SubscribeRequest),
 		maxSubscriberQuality:         make(map[livekit.ParticipantID]*types.SubscribedCodecQuality),
 		maxSubscriberNodeQuality:     make(map[livekit.NodeID][]types.SubscribedCodecQuality),
 		maxSubscribedQuality:         make(map[string]livekit.VideoQuality),
@@ -119,14 +100,14 @@ func (t *MediaTrackSubscriptions) Close() {
 	t.qualityNotifyOpQueue.Stop()
 }
 
-func (t *MediaTrackSubscriptions) OnNoSubscribers(f func()) {
-	t.onNoSubscribers = f
-}
-
 func (t *MediaTrackSubscriptions) OnDownTrackCreated(f func(downTrack *sfu.DownTrack)) {
 	t.onDownTrackCreated = f
 }
 
+func (t *MediaTrackSubscriptions) OnSubscriptionOperationComplete(f func(sub types.LocalParticipant)) {
+	t.onSubscriptionOperationComplete = f
+}
+
 func (t *MediaTrackSubscriptions) SetMuted(muted bool) {
 	// update quality based on subscription if unmuting.
 	// This will queue up the current state, but subscriber
@@ -155,64 +136,8 @@ func (t *MediaTrackSubscriptions) AddCodec(mime string) {
 	t.subscribedTracksMu.Unlock()
 }
 
-func (t *MediaTrackSubscriptions) processRequestsQueue(subscriberID livekit.ParticipantID) {
-	t.subscribedTracksMu.Lock()
-	if t.inProgress[subscriberID] || len(t.requestsQueue[subscriberID]) == 0 {
-		t.subscribedTracksMu.Unlock()
-		return
-	}
-
-	request := t.requestsQueue[subscriberID][0]
-	t.requestsQueue[subscriberID] = t.requestsQueue[subscriberID][1:]
-	if len(t.requestsQueue[subscriberID]) == 0 {
-		delete(t.requestsQueue, subscriberID)
-	}
-
-	t.inProgress[subscriberID] = true
-	t.subscribedTracksMu.Unlock()
-
-	switch request.requestType {
-	case SubscribeRequestTypeAdd:
-		err := t.addSubscriber(request.sub, request.wr)
-		if err != nil {
-			if err != errAlreadySubscribed {
-				t.params.Logger.Errorw("error adding subscriber", err, "subscriberID", subscriberID)
-			}
-
-			// process pending request even if adding errors out
-			go t.clearInProgressAndProcessRequestsQueue(subscriberID)
-		}
-
-	case SubscribeRequestTypeRemove:
-		err := t.removeSubscriber(subscriberID, request.willBeResumed)
-		if err != nil {
-			go t.clearInProgressAndProcessRequestsQueue(subscriberID)
-		}
-
-	default:
-		t.params.Logger.Warnw("unknown request type", nil)
-
-		// let the queue move forward
-		go t.clearInProgressAndProcessRequestsQueue(subscriberID)
-	}
-}
-
 // AddSubscriber subscribes sub to current mediaTrack
 func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *WrappedReceiver) error {
-	subscriberID := sub.ID()
-	t.subscribedTracksMu.Lock()
-	t.requestsQueue[subscriberID] = append(t.requestsQueue[subscriberID], SubscribeRequest{
-		requestType: SubscribeRequestTypeAdd,
-		sub:         sub,
-		wr:          wr,
-	})
-	t.subscribedTracksMu.Unlock()
-
-	t.processRequestsQueue(subscriberID)
-	return nil
-}
-
-func (t *MediaTrackSubscriptions) addSubscriber(sub types.LocalParticipant, wr *WrappedReceiver) error {
 	trackID := t.params.MediaTrack.ID()
 	subscriberID := sub.ID()
 
@@ -254,6 +179,7 @@ func (t *MediaTrackSubscriptions) addSubscriber(sub types.LocalParticipant, wr *
 	subTrack := NewSubscribedTrack(SubscribedTrackParams{
 		PublisherID:       t.params.MediaTrack.PublisherID(),
 		PublisherIdentity: t.params.MediaTrack.PublisherIdentity(),
+		PublisherVersion:  t.params.MediaTrack.PublisherVersion(),
 		Subscriber:        sub,
 		MediaTrack:        t.params.MediaTrack,
 		DownTrack:         downTrack,
@@ -396,7 +322,9 @@ func (t *MediaTrackSubscriptions) addSubscriber(sub types.LocalParticipant, wr *
 			sub.Negotiate(false)
 		}
 
-		t.clearInProgressAndProcessRequestsQueue(subscriberID)
+		if t.onSubscriptionOperationComplete != nil {
+			t.onSubscriptionOperationComplete(sub)
+		}
 	}()
 
 	t.params.Telemetry.TrackSubscribed(
@@ -413,51 +341,17 @@ func (t *MediaTrackSubscriptions) addSubscriber(sub types.LocalParticipant, wr *
 
 // RemoveSubscriber removes participant from subscription
 // stop all forwarders to the client
-func (t *MediaTrackSubscriptions) RemoveSubscriber(subscriberID livekit.ParticipantID, willBeResumed bool) {
-	t.subscribedTracksMu.Lock()
-	t.requestsQueue[subscriberID] = append(t.requestsQueue[subscriberID], SubscribeRequest{
-		requestType:   SubscribeRequestTypeRemove,
-		willBeResumed: willBeResumed,
-	})
-	t.subscribedTracksMu.Unlock()
-
-	t.processRequestsQueue(subscriberID)
-}
-
-func (t *MediaTrackSubscriptions) removeSubscriber(subscriberID livekit.ParticipantID, willBeResumed bool) error {
-	t.params.Logger.Debugw("removing subscriber", "subscriberID", subscriberID, "willBeResumed", willBeResumed)
+func (t *MediaTrackSubscriptions) RemoveSubscriber(subscriberID livekit.ParticipantID, willBeResumed bool) error {
 	subTrack := t.getSubscribedTrack(subscriberID)
 	if subTrack == nil {
 		return errNotFound
 	}
 
+	t.params.Logger.Debugw("removing subscriber", "subscriberID", subscriberID, "willBeResumed", willBeResumed)
 	t.closeSubscribedTrack(subTrack, willBeResumed)
 	return nil
 }
 
-func (t *MediaTrackSubscriptions) RemoveAllSubscribers(willBeResumed bool) {
-	t.params.Logger.Debugw("removing all subscribers")
-
-	var subIDs []livekit.ParticipantID
-	t.subscribedTracksMu.Lock()
-	for _, subTrack := range t.getAllSubscribedTracksLocked() {
-		subscriberID := subTrack.SubscriberID()
-		t.requestsQueue[subscriberID] = append(t.requestsQueue[subscriberID], SubscribeRequest{
-			requestType:   SubscribeRequestTypeRemove,
-			willBeResumed: willBeResumed,
-		})
-
-		subIDs = append(subIDs, subscriberID)
-	}
-	t.subscribedTracksMu.Unlock()
-
-	for _, subID := range subIDs {
-		t.processRequestsQueue(subID)
-	}
-
-	t.maybeNotifyNoSubscribers()
-}
-
 func (t *MediaTrackSubscriptions) closeSubscribedTrack(subTrack types.SubscribedTrack, willBeResumed bool) {
 	dt := subTrack.DownTrack()
 	if dt == nil {
@@ -483,32 +377,6 @@ func (t *MediaTrackSubscriptions) ResyncAllSubscribers() {
 	}
 }
 
-func (t *MediaTrackSubscriptions) RevokeDisallowedSubscribers(allowedSubscriberIdentities []livekit.ParticipantIdentity) []livekit.ParticipantIdentity {
-	var revokedSubscriberIdentities []livekit.ParticipantIdentity
-
-	// LK-TODO: large number of subscribers needs to be solved for this loop
-	for _, subTrack := range t.getAllSubscribedTracks() {
-		found := false
-		for _, allowedIdentity := range allowedSubscriberIdentities {
-			if subTrack.SubscriberIdentity() == allowedIdentity {
-				found = true
-				break
-			}
-		}
-
-		if !found {
-			t.params.Logger.Infow("revoking subscription",
-				"subscriber", subTrack.SubscriberIdentity(),
-				"subscriberID", subTrack.SubscriberID(),
-			)
-			t.RemoveSubscriber(subTrack.SubscriberID(), false)
-			revokedSubscriberIdentities = append(revokedSubscriberIdentities, subTrack.SubscriberIdentity())
-		}
-	}
-
-	return revokedSubscriberIdentities
-}
-
 func (t *MediaTrackSubscriptions) GetAllSubscribers() []livekit.ParticipantID {
 	t.subscribedTracksMu.RLock()
 	defer t.subscribedTracksMu.RUnlock()
@@ -850,26 +718,18 @@ func (t *MediaTrackSubscriptions) stopMaxQualityTimer() {
 	}
 }
 
-func (t *MediaTrackSubscriptions) maybeNotifyNoSubscribers() {
-	if t.onNoSubscribers == nil {
-		return
-	}
-
-	t.subscribedTracksMu.RLock()
-	empty := len(t.subscribedTracks) == 0 && len(t.inProgress) == 0 && len(t.requestsQueue) == 0
-	t.subscribedTracksMu.RUnlock()
-
-	if empty {
-		t.onNoSubscribers()
-	}
-}
-
 func (t *MediaTrackSubscriptions) downTrackClosed(
 	sub types.LocalParticipant,
 	subTrack types.SubscribedTrack,
 	willBeResumed bool,
 	sender *webrtc.RTPSender,
 ) {
+	defer func() {
+		if t.onSubscriptionOperationComplete != nil {
+			t.onSubscriptionOperationComplete(sub)
+		}
+	}()
+
 	subscriberID := sub.ID()
 	t.subscribedTracksMu.Lock()
 	delete(t.subscribedTracks, subscriberID)
@@ -914,15 +774,4 @@ func (t *MediaTrackSubscriptions) downTrackClosed(
 	if !willBeResumed {
 		sub.Negotiate(false)
 	}
-
-	t.clearInProgressAndProcessRequestsQueue(subscriberID)
-	t.maybeNotifyNoSubscribers()
-}
-
-func (t *MediaTrackSubscriptions) clearInProgressAndProcessRequestsQueue(subscriberID livekit.ParticipantID) {
-	t.subscribedTracksMu.Lock()
-	delete(t.inProgress, subscriberID)
-	t.subscribedTracksMu.Unlock()
-
-	t.processRequestsQueue(subscriberID)
 }

pkg/rtc/participant.go

@@ -49,6 +49,20 @@ type downTrackState struct {
 	forwarder   sfu.ForwarderState
 }
 
+type SubscribeRequestType int
+
+const (
+	SubscribeRequestTypeRemove SubscribeRequestType = iota
+	SubscribeRequestTypeAdd
+)
+
+type SubscribeRequest struct {
+	requestType   SubscribeRequestType
+	willBeResumed bool
+	addCb         func(sub types.LocalParticipant) error
+	removeCb      func(subscriberID livekit.ParticipantID, willBeResumed bool) error
+}
+
 type ParticipantParams struct {
 	Identity                livekit.ParticipantIdentity
 	Name                    livekit.ParticipantName
@@ -144,6 +158,10 @@ type ParticipantImpl struct {
 	iceConfig      types.IceConfig
 
 	cachedDownTracks map[livekit.TrackID]*downTrackState
+
+	subscriptionInProgress    map[livekit.TrackID]bool
+	subscriptionRequestsQueue map[livekit.TrackID][]SubscribeRequest
+	trackPublisherVersion     map[livekit.TrackID]uint32
 }
 
 func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
@@ -157,16 +175,19 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
 		return nil, ErrMissingGrants
 	}
 	p := &ParticipantImpl{
-		params:                   params,
-		rtcpCh:                   make(chan []rtcp.Packet, 100),
-		pendingTracks:            make(map[string]*pendingTrackInfo),
-		subscribedTracks:         make(map[livekit.TrackID]types.SubscribedTrack),
-		subscribedTracksSettings: make(map[livekit.TrackID]*livekit.UpdateTrackSettings),
-		disallowedSubscriptions:  make(map[livekit.TrackID]livekit.ParticipantID),
-		subscribedTo:             make(map[livekit.ParticipantID]struct{}),
-		connectedAt:              time.Now(),
-		rttUpdatedAt:             time.Now(),
-		cachedDownTracks:         make(map[livekit.TrackID]*downTrackState),
+		params:                    params,
+		rtcpCh:                    make(chan []rtcp.Packet, 100),
+		pendingTracks:             make(map[string]*pendingTrackInfo),
+		subscribedTracks:          make(map[livekit.TrackID]types.SubscribedTrack),
+		subscribedTracksSettings:  make(map[livekit.TrackID]*livekit.UpdateTrackSettings),
+		disallowedSubscriptions:   make(map[livekit.TrackID]livekit.ParticipantID),
+		subscribedTo:              make(map[livekit.ParticipantID]struct{}),
+		connectedAt:               time.Now(),
+		rttUpdatedAt:              time.Now(),
+		cachedDownTracks:          make(map[livekit.TrackID]*downTrackState),
+		subscriptionInProgress:    make(map[livekit.TrackID]bool),
+		subscriptionRequestsQueue: make(map[livekit.TrackID][]SubscribeRequest),
+		trackPublisherVersion:     make(map[livekit.TrackID]uint32),
 	}
 	p.version.Store(params.InitialVersion)
 	p.migrateState.Store(types.MigrateStateInit)
@@ -954,8 +975,17 @@ func (p *ParticipantImpl) AddSubscribedTrack(subTrack types.SubscribedTrack) {
 		"publisherIdentity", subTrack.PublisherIdentity(),
 		"trackID", subTrack.ID())
 	p.lock.Lock()
+	if v, ok := p.trackPublisherVersion[subTrack.ID()]; ok && v > subTrack.PublisherVersion() {
+		p.lock.Unlock()
+		p.params.Logger.Infow("ignoring add subscribedTrack from older version", "current", v, "requesting", subTrack.PublisherVersion())
+		return
+	}
+	p.trackPublisherVersion[subTrack.ID()] = subTrack.PublisherVersion()
+
 	onSubscribedTo := p.onSubscribedTo
+
 	p.subscribedTracks[subTrack.ID()] = subTrack
+
 	settings := p.subscribedTracksSettings[subTrack.ID()]
 	p.lock.Unlock()
 
@@ -987,10 +1017,16 @@ func (p *ParticipantImpl) RemoveSubscribedTrack(subTrack types.SubscribedTrack)
 		"publisherIdentity", subTrack.PublisherIdentity(),
 		"trackID", subTrack.ID(), "kind", subTrack.DownTrack().Kind())
 
-	p.subscriber.RemoveTrack(subTrack)
-
 	p.lock.Lock()
+	if v, ok := p.trackPublisherVersion[subTrack.ID()]; ok && v > subTrack.PublisherVersion() {
+		p.lock.Unlock()
+		p.params.Logger.Infow("ignoring remove subscribedTrack from older version", "current", v, "requesting", subTrack.PublisherVersion())
+		return
+	}
+	p.trackPublisherVersion[subTrack.ID()] = subTrack.PublisherVersion()
+
 	delete(p.subscribedTracks, subTrack.ID())
+
 	// remove from subscribed map
 	numRemaining := 0
 	for _, st := range p.subscribedTracks {
@@ -1010,6 +1046,8 @@ func (p *ParticipantImpl) RemoveSubscribedTrack(subTrack types.SubscribedTrack)
 	}
 	p.lock.Unlock()
 
+	p.subscriber.RemoveTrack(subTrack)
+
 	if numRemaining == 0 {
 		//
 		// When a participant leaves OR
@@ -1659,6 +1697,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei
 			SdpCid:              track.ID(),
 			ParticipantID:       p.params.SID,
 			ParticipantIdentity: p.params.Identity,
+			ParticipantVersion:  p.version.Load(),
 			RTCPChan:            p.rtcpCh,
 			BufferFactory:       p.params.Config.BufferFactory,
 			ReceiverConfig:      p.params.Config.Receiver,
@@ -1723,6 +1762,7 @@ func (p *ParticipantImpl) addMigrateMutedTrack(cid string, t *livekit.TrackInfo)
 		SdpCid:              cid,
 		ParticipantID:       p.params.SID,
 		ParticipantIdentity: p.params.Identity,
+		ParticipantVersion:  p.version.Load(),
 		RTCPChan:            p.rtcpCh,
 		BufferFactory:       p.params.Config.BufferFactory,
 		ReceiverConfig:      p.params.Config.Receiver,
@@ -2070,3 +2110,80 @@ func (p *ParticipantImpl) handleNegotiationFailed() {
 	})
 	p.closeSignalConnection()
 }
+
+func (p *ParticipantImpl) EnqueueSubscribeTrack(trackID livekit.TrackID, f func(sub types.LocalParticipant) error) {
+	p.params.Logger.Infow("queueing subscribe", "trackID", trackID)
+
+	p.lock.Lock()
+	p.subscriptionRequestsQueue[trackID] = append(p.subscriptionRequestsQueue[trackID], SubscribeRequest{
+		requestType: SubscribeRequestTypeAdd,
+		addCb:       f,
+	})
+	p.lock.Unlock()
+
+	go p.ProcessSubscriptionRequestsQueue(trackID)
+}
+
+func (p *ParticipantImpl) EnqueueUnsubscribeTrack(trackID livekit.TrackID, willBeResumed bool, f func(subscriberID livekit.ParticipantID, willBeResumed bool) error) {
+	p.params.Logger.Infow("queueing unsubscribe", "trackID", trackID)
+
+	p.lock.Lock()
+	p.subscriptionRequestsQueue[trackID] = append(p.subscriptionRequestsQueue[trackID], SubscribeRequest{
+		requestType:   SubscribeRequestTypeRemove,
+		willBeResumed: willBeResumed,
+		removeCb:      f,
+	})
+	p.lock.Unlock()
+
+	go p.ProcessSubscriptionRequestsQueue(trackID)
+}
+
+func (p *ParticipantImpl) ProcessSubscriptionRequestsQueue(trackID livekit.TrackID) {
+	p.lock.Lock()
+	if p.subscriptionInProgress[trackID] || len(p.subscriptionRequestsQueue[trackID]) == 0 {
+		p.lock.Unlock()
+		return
+	}
+
+	request := p.subscriptionRequestsQueue[trackID][0]
+	p.subscriptionRequestsQueue[trackID] = p.subscriptionRequestsQueue[trackID][1:]
+	if len(p.subscriptionRequestsQueue[trackID]) == 0 {
+		delete(p.subscriptionRequestsQueue, trackID)
+	}
+
+	p.subscriptionInProgress[trackID] = true
+	p.lock.Unlock()
+
+	switch request.requestType {
+	case SubscribeRequestTypeAdd:
+		err := request.addCb(p)
+		if err != nil {
+			if err != errAlreadySubscribed {
+				p.params.Logger.Errorw("error adding subscriber", err, "trackID", trackID)
+			}
+
+			// process pending request even if adding errors out
+			go p.ClearInProgressAndProcessSubscriptionRequestsQueue(trackID)
+		}
+
+	case SubscribeRequestTypeRemove:
+		err := request.removeCb(p.ID(), request.willBeResumed)
+		if err != nil {
+			go p.ClearInProgressAndProcessSubscriptionRequestsQueue(trackID)
+		}
+
+	default:
+		p.params.Logger.Warnw("unknown request type", nil)
+
+		// let the queue move forward
+		go p.ClearInProgressAndProcessSubscriptionRequestsQueue(trackID)
+	}
+}
+
+func (p *ParticipantImpl) ClearInProgressAndProcessSubscriptionRequestsQueue(trackID livekit.TrackID) {
+	p.lock.Lock()
+	delete(p.subscriptionInProgress, trackID)
+	p.lock.Unlock()
+
+	p.ProcessSubscriptionRequestsQueue(trackID)
+}

pkg/rtc/subscribedtrack.go

@@ -21,6 +21,7 @@ const (
 type SubscribedTrackParams struct {
 	PublisherID       livekit.ParticipantID
 	PublisherIdentity livekit.ParticipantIdentity
+	PublisherVersion  uint32
 	Subscriber        types.LocalParticipant
 	MediaTrack        types.MediaTrack
 	DownTrack         *sfu.DownTrack
@@ -80,6 +81,10 @@ func (t *SubscribedTrack) PublisherIdentity() livekit.ParticipantIdentity {
 	return t.params.PublisherIdentity
 }
 
+func (t *SubscribedTrack) PublisherVersion() uint32 {
+	return t.params.PublisherVersion
+}
+
 func (t *SubscribedTrack) SubscriberID() livekit.ParticipantID {
 	return t.params.Subscriber.ID()
 }

pkg/rtc/transport.go

@@ -267,6 +267,13 @@ func (t *PCTransport) IsEstablished() bool {
 }
 
 func (t *PCTransport) Close() {
+	t.lock.Lock()
+	if t.signalStateCheckTimer != nil {
+		t.signalStateCheckTimer.Stop()
+		t.signalStateCheckTimer = nil
+	}
+	t.lock.Unlock()
+
 	if t.streamAllocator != nil {
 		t.streamAllocator.Stop()
 	}
@@ -288,6 +295,7 @@ func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error {
 
 	if t.signalStateCheckTimer != nil {
 		t.signalStateCheckTimer.Stop()
+		t.signalStateCheckTimer = nil
 	}
 
 	for _, c := range t.pendingCandidates {
@@ -434,6 +442,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error {
 	negotiateVersion := t.negotiateCounter.Inc()
 	if t.signalStateCheckTimer != nil {
 		t.signalStateCheckTimer.Stop()
+		t.signalStateCheckTimer = nil
 	}
 	t.signalStateCheckTimer = time.AfterFunc(negotiationFailedTimout, func() {
 		t.lock.RLock()

pkg/rtc/types/interfaces.go

@@ -284,6 +284,11 @@ type LocalParticipant interface {
 	CacheDownTrack(trackID livekit.TrackID, rtpTransceiver *webrtc.RTPTransceiver, forwarderState sfu.ForwarderState)
 	UncacheDownTrack(rtpTransceiver *webrtc.RTPTransceiver)
 	GetCachedDownTrack(trackID livekit.TrackID) (*webrtc.RTPTransceiver, sfu.ForwarderState)
+
+	EnqueueSubscribeTrack(trackID livekit.TrackID, f func(sub LocalParticipant) error)
+	EnqueueUnsubscribeTrack(trackID livekit.TrackID, willBeResumed bool, f func(subscriberID livekit.ParticipantID, willBeResumed bool) error)
+	ProcessSubscriptionRequestsQueue(trackID livekit.TrackID)
+	ClearInProgressAndProcessSubscriptionRequestsQueue(trackID livekit.TrackID)
 }
 
 // Room is a container of participants, and can provide room-level actions
@@ -312,6 +317,7 @@ type MediaTrack interface {
 
 	PublisherID() livekit.ParticipantID
 	PublisherIdentity() livekit.ParticipantIdentity
+	PublisherVersion() uint32
 
 	IsMuted() bool
 	SetMuted(muted bool)
@@ -362,6 +368,7 @@ type SubscribedTrack interface {
 	ID() livekit.TrackID
 	PublisherID() livekit.ParticipantID
 	PublisherIdentity() livekit.ParticipantIdentity
+	PublisherVersion() uint32
 	SubscriberID() livekit.ParticipantID
 	SubscriberIdentity() livekit.ParticipantIdentity
 	Subscriber() LocalParticipant

pkg/rtc/types/typesfakes/fake_local_media_track.go

@@ -184,6 +184,16 @@ type FakeLocalMediaTrack struct {
 	publisherIdentityReturnsOnCall map[int]struct {
 		result1 livekit.ParticipantIdentity
 	}
+	PublisherVersionStub        func() uint32
+	publisherVersionMutex       sync.RWMutex
+	publisherVersionArgsForCall []struct {
+	}
+	publisherVersionReturns struct {
+		result1 uint32
+	}
+	publisherVersionReturnsOnCall map[int]struct {
+		result1 uint32
+	}
 	ReceiversStub        func() []sfu.TrackReceiver
 	receiversMutex       sync.RWMutex
 	receiversArgsForCall []struct {
@@ -1203,6 +1213,59 @@ func (fake *FakeLocalMediaTrack) PublisherIdentityReturnsOnCall(i int, result1 l
 	}{result1}
 }
 
+func (fake *FakeLocalMediaTrack) PublisherVersion() uint32 {
+	fake.publisherVersionMutex.Lock()
+	ret, specificReturn := fake.publisherVersionReturnsOnCall[len(fake.publisherVersionArgsForCall)]
+	fake.publisherVersionArgsForCall = append(fake.publisherVersionArgsForCall, struct {
+	}{})
+	stub := fake.PublisherVersionStub
+	fakeReturns := fake.publisherVersionReturns
+	fake.recordInvocation("PublisherVersion", []interface{}{})
+	fake.publisherVersionMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeLocalMediaTrack) PublisherVersionCallCount() int {
+	fake.publisherVersionMutex.RLock()
+	defer fake.publisherVersionMutex.RUnlock()
+	return len(fake.publisherVersionArgsForCall)
+}
+
+func (fake *FakeLocalMediaTrack) PublisherVersionCalls(stub func() uint32) {
+	fake.publisherVersionMutex.Lock()
+	defer fake.publisherVersionMutex.Unlock()
+	fake.PublisherVersionStub = stub
+}
+
+func (fake *FakeLocalMediaTrack) PublisherVersionReturns(result1 uint32) {
+	fake.publisherVersionMutex.Lock()
+	defer fake.publisherVersionMutex.Unlock()
+	fake.PublisherVersionStub = nil
+	fake.publisherVersionReturns = struct {
+		result1 uint32
+	}{result1}
+}
+
+func (fake *FakeLocalMediaTrack) PublisherVersionReturnsOnCall(i int, result1 uint32) {
+	fake.publisherVersionMutex.Lock()
+	defer fake.publisherVersionMutex.Unlock()
+	fake.PublisherVersionStub = nil
+	if fake.publisherVersionReturnsOnCall == nil {
+		fake.publisherVersionReturnsOnCall = make(map[int]struct {
+			result1 uint32
+		})
+	}
+	fake.publisherVersionReturnsOnCall[i] = struct {
+		result1 uint32
+	}{result1}
+}
+
 func (fake *FakeLocalMediaTrack) Receivers() []sfu.TrackReceiver {
 	fake.receiversMutex.Lock()
 	ret, specificReturn := fake.receiversReturnsOnCall[len(fake.receiversArgsForCall)]
@@ -1710,6 +1773,8 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} {
 	defer fake.publisherIDMutex.RUnlock()
 	fake.publisherIdentityMutex.RLock()
 	defer fake.publisherIdentityMutex.RUnlock()
+	fake.publisherVersionMutex.RLock()
+	defer fake.publisherVersionMutex.RUnlock()
 	fake.receiversMutex.RLock()
 	defer fake.receiversMutex.RUnlock()
 	fake.removeAllSubscribersMutex.RLock()

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -98,6 +98,11 @@ type FakeLocalParticipant struct {
 	claimGrantsReturnsOnCall map[int]struct {
 		result1 *auth.ClaimGrants
 	}
+	ClearInProgressAndProcessSubscriptionRequestsQueueStub        func(livekit.TrackID)
+	clearInProgressAndProcessSubscriptionRequestsQueueMutex       sync.RWMutex
+	clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall []struct {
+		arg1 livekit.TrackID
+	}
 	CloseStub        func(bool, types.ParticipantCloseReason) error
 	closeMutex       sync.RWMutex
 	closeArgsForCall []struct {
@@ -130,6 +135,19 @@ type FakeLocalParticipant struct {
 	debugInfoReturnsOnCall map[int]struct {
 		result1 map[string]interface{}
 	}
+	EnqueueSubscribeTrackStub        func(livekit.TrackID, func(sub types.LocalParticipant) error)
+	enqueueSubscribeTrackMutex       sync.RWMutex
+	enqueueSubscribeTrackArgsForCall []struct {
+		arg1 livekit.TrackID
+		arg2 func(sub types.LocalParticipant) error
+	}
+	EnqueueUnsubscribeTrackStub        func(livekit.TrackID, bool, func(subscriberID livekit.ParticipantID, willBeResumed bool) error)
+	enqueueUnsubscribeTrackMutex       sync.RWMutex
+	enqueueUnsubscribeTrackArgsForCall []struct {
+		arg1 livekit.TrackID
+		arg2 bool
+		arg3 func(subscriberID livekit.ParticipantID, willBeResumed bool) error
+	}
 	GetAdaptiveStreamStub        func() bool
 	getAdaptiveStreamMutex       sync.RWMutex
 	getAdaptiveStreamArgsForCall []struct {
@@ -397,6 +415,11 @@ type FakeLocalParticipant struct {
 	onTrackUpdatedArgsForCall []struct {
 		arg1 func(types.LocalParticipant, types.MediaTrack)
 	}
+	ProcessSubscriptionRequestsQueueStub        func(livekit.TrackID)
+	processSubscriptionRequestsQueueMutex       sync.RWMutex
+	processSubscriptionRequestsQueueArgsForCall []struct {
+		arg1 livekit.TrackID
+	}
 	ProtocolVersionStub        func() types.ProtocolVersion
 	protocolVersionMutex       sync.RWMutex
 	protocolVersionArgsForCall []struct {
@@ -1123,6 +1146,38 @@ func (fake *FakeLocalParticipant) ClaimGrantsReturnsOnCall(i int, result1 *auth.
 	}{result1}
 }
 
+func (fake *FakeLocalParticipant) ClearInProgressAndProcessSubscriptionRequestsQueue(arg1 livekit.TrackID) {
+	fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.Lock()
+	fake.clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall = append(fake.clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall, struct {
+		arg1 livekit.TrackID
+	}{arg1})
+	stub := fake.ClearInProgressAndProcessSubscriptionRequestsQueueStub
+	fake.recordInvocation("ClearInProgressAndProcessSubscriptionRequestsQueue", []interface{}{arg1})
+	fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.Unlock()
+	if stub != nil {
+		fake.ClearInProgressAndProcessSubscriptionRequestsQueueStub(arg1)
+	}
+}
+
+func (fake *FakeLocalParticipant) ClearInProgressAndProcessSubscriptionRequestsQueueCallCount() int {
+	fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RLock()
+	defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RUnlock()
+	return len(fake.clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) ClearInProgressAndProcessSubscriptionRequestsQueueCalls(stub func(livekit.TrackID)) {
+	fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.Lock()
+	defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.Unlock()
+	fake.ClearInProgressAndProcessSubscriptionRequestsQueueStub = stub
+}
+
+func (fake *FakeLocalParticipant) ClearInProgressAndProcessSubscriptionRequestsQueueArgsForCall(i int) livekit.TrackID {
+	fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RLock()
+	defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RUnlock()
+	argsForCall := fake.clearInProgressAndProcessSubscriptionRequestsQueueArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakeLocalParticipant) Close(arg1 bool, arg2 types.ParticipantCloseReason) error {
 	fake.closeMutex.Lock()
 	ret, specificReturn := fake.closeReturnsOnCall[len(fake.closeArgsForCall)]
@@ -1291,6 +1346,73 @@ func (fake *FakeLocalParticipant) DebugInfoReturnsOnCall(i int, result1 map[stri
 	}{result1}
 }
 
+func (fake *FakeLocalParticipant) EnqueueSubscribeTrack(arg1 livekit.TrackID, arg2 func(sub types.LocalParticipant) error) {
+	fake.enqueueSubscribeTrackMutex.Lock()
+	fake.enqueueSubscribeTrackArgsForCall = append(fake.enqueueSubscribeTrackArgsForCall, struct {
+		arg1 livekit.TrackID
+		arg2 func(sub types.LocalParticipant) error
+	}{arg1, arg2})
+	stub := fake.EnqueueSubscribeTrackStub
+	fake.recordInvocation("EnqueueSubscribeTrack", []interface{}{arg1, arg2})
+	fake.enqueueSubscribeTrackMutex.Unlock()
+	if stub != nil {
+		fake.EnqueueSubscribeTrackStub(arg1, arg2)
+	}
+}
+
+func (fake *FakeLocalParticipant) EnqueueSubscribeTrackCallCount() int {
+	fake.enqueueSubscribeTrackMutex.RLock()
+	defer fake.enqueueSubscribeTrackMutex.RUnlock()
+	return len(fake.enqueueSubscribeTrackArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) EnqueueSubscribeTrackCalls(stub func(livekit.TrackID, func(sub types.LocalParticipant) error)) {
+	fake.enqueueSubscribeTrackMutex.Lock()
+	defer fake.enqueueSubscribeTrackMutex.Unlock()
+	fake.EnqueueSubscribeTrackStub = stub
+}
+
+func (fake *FakeLocalParticipant) EnqueueSubscribeTrackArgsForCall(i int) (livekit.TrackID, func(sub types.LocalParticipant) error) {
+	fake.enqueueSubscribeTrackMutex.RLock()
+	defer fake.enqueueSubscribeTrackMutex.RUnlock()
+	argsForCall := fake.enqueueSubscribeTrackArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *FakeLocalParticipant) EnqueueUnsubscribeTrack(arg1 livekit.TrackID, arg2 bool, arg3 func(subscriberID livekit.ParticipantID, willBeResumed bool) error) {
+	fake.enqueueUnsubscribeTrackMutex.Lock()
+	fake.enqueueUnsubscribeTrackArgsForCall = append(fake.enqueueUnsubscribeTrackArgsForCall, struct {
+		arg1 livekit.TrackID
+		arg2 bool
+		arg3 func(subscriberID livekit.ParticipantID, willBeResumed bool) error
+	}{arg1, arg2, arg3})
+	stub := fake.EnqueueUnsubscribeTrackStub
+	fake.recordInvocation("EnqueueUnsubscribeTrack", []interface{}{arg1, arg2, arg3})
+	fake.enqueueUnsubscribeTrackMutex.Unlock()
+	if stub != nil {
+		fake.EnqueueUnsubscribeTrackStub(arg1, arg2, arg3)
+	}
+}
+
+func (fake *FakeLocalParticipant) EnqueueUnsubscribeTrackCallCount() int {
+	fake.enqueueUnsubscribeTrackMutex.RLock()
+	defer fake.enqueueUnsubscribeTrackMutex.RUnlock()
+	return len(fake.enqueueUnsubscribeTrackArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) EnqueueUnsubscribeTrackCalls(stub func(livekit.TrackID, bool, func(subscriberID livekit.ParticipantID, willBeResumed bool) error)) {
+	fake.enqueueUnsubscribeTrackMutex.Lock()
+	defer fake.enqueueUnsubscribeTrackMutex.Unlock()
+	fake.EnqueueUnsubscribeTrackStub = stub
+}
+
+func (fake *FakeLocalParticipant) EnqueueUnsubscribeTrackArgsForCall(i int) (livekit.TrackID, bool, func(subscriberID livekit.ParticipantID, willBeResumed bool) error) {
+	fake.enqueueUnsubscribeTrackMutex.RLock()
+	defer fake.enqueueUnsubscribeTrackMutex.RUnlock()
+	argsForCall := fake.enqueueUnsubscribeTrackArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
+}
+
 func (fake *FakeLocalParticipant) GetAdaptiveStream() bool {
 	fake.getAdaptiveStreamMutex.Lock()
 	ret, specificReturn := fake.getAdaptiveStreamReturnsOnCall[len(fake.getAdaptiveStreamArgsForCall)]
@@ -2749,6 +2871,38 @@ func (fake *FakeLocalParticipant) OnTrackUpdatedArgsForCall(i int) func(types.Lo
 	return argsForCall.arg1
 }
 
+func (fake *FakeLocalParticipant) ProcessSubscriptionRequestsQueue(arg1 livekit.TrackID) {
+	fake.processSubscriptionRequestsQueueMutex.Lock()
+	fake.processSubscriptionRequestsQueueArgsForCall = append(fake.processSubscriptionRequestsQueueArgsForCall, struct {
+		arg1 livekit.TrackID
+	}{arg1})
+	stub := fake.ProcessSubscriptionRequestsQueueStub
+	fake.recordInvocation("ProcessSubscriptionRequestsQueue", []interface{}{arg1})
+	fake.processSubscriptionRequestsQueueMutex.Unlock()
+	if stub != nil {
+		fake.ProcessSubscriptionRequestsQueueStub(arg1)
+	}
+}
+
+func (fake *FakeLocalParticipant) ProcessSubscriptionRequestsQueueCallCount() int {
+	fake.processSubscriptionRequestsQueueMutex.RLock()
+	defer fake.processSubscriptionRequestsQueueMutex.RUnlock()
+	return len(fake.processSubscriptionRequestsQueueArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) ProcessSubscriptionRequestsQueueCalls(stub func(livekit.TrackID)) {
+	fake.processSubscriptionRequestsQueueMutex.Lock()
+	defer fake.processSubscriptionRequestsQueueMutex.Unlock()
+	fake.ProcessSubscriptionRequestsQueueStub = stub
+}
+
+func (fake *FakeLocalParticipant) ProcessSubscriptionRequestsQueueArgsForCall(i int) livekit.TrackID {
+	fake.processSubscriptionRequestsQueueMutex.RLock()
+	defer fake.processSubscriptionRequestsQueueMutex.RUnlock()
+	argsForCall := fake.processSubscriptionRequestsQueueArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakeLocalParticipant) ProtocolVersion() types.ProtocolVersion {
 	fake.protocolVersionMutex.Lock()
 	ret, specificReturn := fake.protocolVersionReturnsOnCall[len(fake.protocolVersionArgsForCall)]
@@ -4331,12 +4485,18 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.canSubscribeMutex.RUnlock()
 	fake.claimGrantsMutex.RLock()
 	defer fake.claimGrantsMutex.RUnlock()
+	fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RLock()
+	defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RUnlock()
 	fake.closeMutex.RLock()
 	defer fake.closeMutex.RUnlock()
 	fake.connectedAtMutex.RLock()
 	defer fake.connectedAtMutex.RUnlock()
 	fake.debugInfoMutex.RLock()
 	defer fake.debugInfoMutex.RUnlock()
+	fake.enqueueSubscribeTrackMutex.RLock()
+	defer fake.enqueueSubscribeTrackMutex.RUnlock()
+	fake.enqueueUnsubscribeTrackMutex.RLock()
+	defer fake.enqueueUnsubscribeTrackMutex.RUnlock()
 	fake.getAdaptiveStreamMutex.RLock()
 	defer fake.getAdaptiveStreamMutex.RUnlock()
 	fake.getAudioLevelMutex.RLock()
@@ -4397,6 +4557,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.onTrackPublishedMutex.RUnlock()
 	fake.onTrackUpdatedMutex.RLock()
 	defer fake.onTrackUpdatedMutex.RUnlock()
+	fake.processSubscriptionRequestsQueueMutex.RLock()
+	defer fake.processSubscriptionRequestsQueueMutex.RUnlock()
 	fake.protocolVersionMutex.RLock()
 	defer fake.protocolVersionMutex.RUnlock()
 	fake.removeSubscribedTrackMutex.RLock()

pkg/rtc/types/typesfakes/fake_media_track.go

@@ -151,6 +151,16 @@ type FakeMediaTrack struct {
 	publisherIdentityReturnsOnCall map[int]struct {
 		result1 livekit.ParticipantIdentity
 	}
+	PublisherVersionStub        func() uint32
+	publisherVersionMutex       sync.RWMutex
+	publisherVersionArgsForCall []struct {
+	}
+	publisherVersionReturns struct {
+		result1 uint32
+	}
+	publisherVersionReturnsOnCall map[int]struct {
+		result1 uint32
+	}
 	ReceiversStub        func() []sfu.TrackReceiver
 	receiversMutex       sync.RWMutex
 	receiversArgsForCall []struct {
@@ -985,6 +995,59 @@ func (fake *FakeMediaTrack) PublisherIdentityReturnsOnCall(i int, result1 liveki
 	}{result1}
 }
 
+func (fake *FakeMediaTrack) PublisherVersion() uint32 {
+	fake.publisherVersionMutex.Lock()
+	ret, specificReturn := fake.publisherVersionReturnsOnCall[len(fake.publisherVersionArgsForCall)]
+	fake.publisherVersionArgsForCall = append(fake.publisherVersionArgsForCall, struct {
+	}{})
+	stub := fake.PublisherVersionStub
+	fakeReturns := fake.publisherVersionReturns
+	fake.recordInvocation("PublisherVersion", []interface{}{})
+	fake.publisherVersionMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeMediaTrack) PublisherVersionCallCount() int {
+	fake.publisherVersionMutex.RLock()
+	defer fake.publisherVersionMutex.RUnlock()
+	return len(fake.publisherVersionArgsForCall)
+}
+
+func (fake *FakeMediaTrack) PublisherVersionCalls(stub func() uint32) {
+	fake.publisherVersionMutex.Lock()
+	defer fake.publisherVersionMutex.Unlock()
+	fake.PublisherVersionStub = stub
+}
+
+func (fake *FakeMediaTrack) PublisherVersionReturns(result1 uint32) {
+	fake.publisherVersionMutex.Lock()
+	defer fake.publisherVersionMutex.Unlock()
+	fake.PublisherVersionStub = nil
+	fake.publisherVersionReturns = struct {
+		result1 uint32
+	}{result1}
+}
+
+func (fake *FakeMediaTrack) PublisherVersionReturnsOnCall(i int, result1 uint32) {
+	fake.publisherVersionMutex.Lock()
+	defer fake.publisherVersionMutex.Unlock()
+	fake.PublisherVersionStub = nil
+	if fake.publisherVersionReturnsOnCall == nil {
+		fake.publisherVersionReturnsOnCall = make(map[int]struct {
+			result1 uint32
+		})
+	}
+	fake.publisherVersionReturnsOnCall[i] = struct {
+		result1 uint32
+	}{result1}
+}
+
 func (fake *FakeMediaTrack) Receivers() []sfu.TrackReceiver {
 	fake.receiversMutex.Lock()
 	ret, specificReturn := fake.receiversReturnsOnCall[len(fake.receiversArgsForCall)]
@@ -1401,6 +1464,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} {
 	defer fake.publisherIDMutex.RUnlock()
 	fake.publisherIdentityMutex.RLock()
 	defer fake.publisherIdentityMutex.RUnlock()
+	fake.publisherVersionMutex.RLock()
+	defer fake.publisherVersionMutex.RUnlock()
 	fake.receiversMutex.RLock()
 	defer fake.receiversMutex.RUnlock()
 	fake.removeAllSubscribersMutex.RLock()

pkg/rtc/types/typesfakes/fake_subscribed_track.go

@@ -75,6 +75,16 @@ type FakeSubscribedTrack struct {
 	publisherIdentityReturnsOnCall map[int]struct {
 		result1 livekit.ParticipantIdentity
 	}
+	PublisherVersionStub        func() uint32
+	publisherVersionMutex       sync.RWMutex
+	publisherVersionArgsForCall []struct {
+	}
+	publisherVersionReturns struct {
+		result1 uint32
+	}
+	publisherVersionReturnsOnCall map[int]struct {
+		result1 uint32
+	}
 	SetPublisherMutedStub        func(bool)
 	setPublisherMutedMutex       sync.RWMutex
 	setPublisherMutedArgsForCall []struct {
@@ -473,6 +483,59 @@ func (fake *FakeSubscribedTrack) PublisherIdentityReturnsOnCall(i int, result1 l
 	}{result1}
 }
 
+func (fake *FakeSubscribedTrack) PublisherVersion() uint32 {
+	fake.publisherVersionMutex.Lock()
+	ret, specificReturn := fake.publisherVersionReturnsOnCall[len(fake.publisherVersionArgsForCall)]
+	fake.publisherVersionArgsForCall = append(fake.publisherVersionArgsForCall, struct {
+	}{})
+	stub := fake.PublisherVersionStub
+	fakeReturns := fake.publisherVersionReturns
+	fake.recordInvocation("PublisherVersion", []interface{}{})
+	fake.publisherVersionMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeSubscribedTrack) PublisherVersionCallCount() int {
+	fake.publisherVersionMutex.RLock()
+	defer fake.publisherVersionMutex.RUnlock()
+	return len(fake.publisherVersionArgsForCall)
+}
+
+func (fake *FakeSubscribedTrack) PublisherVersionCalls(stub func() uint32) {
+	fake.publisherVersionMutex.Lock()
+	defer fake.publisherVersionMutex.Unlock()
+	fake.PublisherVersionStub = stub
+}
+
+func (fake *FakeSubscribedTrack) PublisherVersionReturns(result1 uint32) {
+	fake.publisherVersionMutex.Lock()
+	defer fake.publisherVersionMutex.Unlock()
+	fake.PublisherVersionStub = nil
+	fake.publisherVersionReturns = struct {
+		result1 uint32
+	}{result1}
+}
+
+func (fake *FakeSubscribedTrack) PublisherVersionReturnsOnCall(i int, result1 uint32) {
+	fake.publisherVersionMutex.Lock()
+	defer fake.publisherVersionMutex.Unlock()
+	fake.PublisherVersionStub = nil
+	if fake.publisherVersionReturnsOnCall == nil {
+		fake.publisherVersionReturnsOnCall = make(map[int]struct {
+			result1 uint32
+		})
+	}
+	fake.publisherVersionReturnsOnCall[i] = struct {
+		result1 uint32
+	}{result1}
+}
+
 func (fake *FakeSubscribedTrack) SetPublisherMuted(arg1 bool) {
 	fake.setPublisherMutedMutex.Lock()
 	fake.setPublisherMutedArgsForCall = append(fake.setPublisherMutedArgsForCall, struct {
@@ -737,6 +800,8 @@ func (fake *FakeSubscribedTrack) Invocations() map[string][][]interface{} {
 	defer fake.publisherIDMutex.RUnlock()
 	fake.publisherIdentityMutex.RLock()
 	defer fake.publisherIdentityMutex.RUnlock()
+	fake.publisherVersionMutex.RLock()
+	defer fake.publisherVersionMutex.RUnlock()
 	fake.setPublisherMutedMutex.RLock()
 	defer fake.setPublisherMutedMutex.RUnlock()
 	fake.subscriberMutex.RLock()