Refactor media track subscriptions (#314)

* Refactor media track subscriptions

- To enable re-use of common bits
- Add max quality from other nodes

* Lock close handlers slice

* Reverting multiple on close handlers of downtrack, unclear if it is needed yet

* Make Logger a pointer

* audio level in MediaTrack like remote media track

* Cleanup

* Add a no subscribers callback

* Add method to update subscribed quality from another node

* loss proxying from remote node

* Address comments from David

* create subscriber node quality map

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/google/wire v0.5.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/hashicorp/golang-lru v0.5.4
-	github.com/livekit/protocol v0.11.8-0.20220103045453-c441eb5f03c8
+	github.com/livekit/protocol v0.11.8-0.20220104065946-2c4c8d7764ed
 	github.com/magefile/mage v1.11.0
 	github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
 	github.com/mitchellh/go-homedir v1.1.0

go.sum

@@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA=
 github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
-github.com/livekit/protocol v0.11.8-0.20220103045453-c441eb5f03c8 h1:eo4OOUKLgNguaHEmcItTfk2IzVC2s2r6/WXqFfj5HjE=
-github.com/livekit/protocol v0.11.8-0.20220103045453-c441eb5f03c8/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
+github.com/livekit/protocol v0.11.8-0.20220104065946-2c4c8d7764ed h1:6vxJ62pwuhXtEjqvsANTIoEcTgHR9laMa9tR3Xr0fAM=
+github.com/livekit/protocol v0.11.8-0.20220104065946-2c4c8d7764ed/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
 github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls=
 github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
 github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=

pkg/rtc/mediatrack.go

@@ -8,6 +8,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
 
 	"github.com/livekit/protocol/livekit"
@@ -15,10 +16,8 @@ import (
 	"github.com/livekit/protocol/utils"
 	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
-	"github.com/pion/webrtc/v3/pkg/rtcerr"
 
 	"github.com/livekit/livekit-server/pkg/config"
-	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/livekit-server/pkg/sfu/twcc"
@@ -29,7 +28,6 @@ const (
 	lostUpdateDelta                 = time.Second
 	connectionQualityUpdateInterval = 5 * time.Second
 	layerSelectionTolerance         = 0.9
-	initialQualityUpdateWait        = 10 * time.Second
 )
 
 // MediaTrack represents a WebRTC track that needs to be forwarded
@@ -46,13 +44,13 @@ type MediaTrack struct {
 
 	lock sync.RWMutex
 
-	// map of target participantID -> types.SubscribedTrack
-	subscribedTracks sync.Map // participantID => types.SubscribedTrack
-	twcc             *twcc.Responder
-	audioLevel       *AudioLevel
-	receiver         sfu.Receiver
-	lastPLI          time.Time
-	layerDimensions  sync.Map // quality => *livekit.VideoLayer
+	twcc *twcc.Responder
+
+	audioLevelMu sync.RWMutex
+	audioLevel   *AudioLevel
+
+	receiver        sfu.Receiver
+	layerDimensions sync.Map // livekit.VideoQuality => *livekit.VideoLayer
 
 	// track audio fraction lost
 	statsLock         sync.Mutex
@@ -65,15 +63,9 @@ type MediaTrack struct {
 
 	done chan struct{}
 
-	// quality level enable/disable
-	maxQualityLock               sync.RWMutex
-	maxSubscriberQuality         map[livekit.ParticipantID]livekit.VideoQuality
-	maxSubscribedQuality         livekit.VideoQuality
-	allSubscribersMuted          bool
-	onSubscribedMaxQualityChange func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality) error
-	maxQualityTimer              *time.Timer
-
 	onClose []func()
+
+	*MediaTrackSubscriptions
 }
 
 type MediaTrackParams struct {
@@ -95,15 +87,23 @@ type MediaTrackParams struct {
 
 func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTrack {
 	t := &MediaTrack{
-		params:               params,
-		ssrc:                 track.SSRC(),
-		streamID:             track.StreamID(),
-		codec:                track.Codec(),
-		connectionStats:      connectionquality.NewConnectionStats(),
-		done:                 make(chan struct{}),
-		maxSubscriberQuality: make(map[livekit.ParticipantID]livekit.VideoQuality),
+		params:          params,
+		ssrc:            track.SSRC(),
+		streamID:        track.StreamID(),
+		codec:           track.Codec(),
+		connectionStats: connectionquality.NewConnectionStats(),
+		done:            make(chan struct{}),
 	}
 
+	t.MediaTrackSubscriptions = NewMediaTrackSubscriptions(MediaTrackSubscriptionsParams{
+		MediaTrack:       t,
+		BufferFactory:    params.BufferFactory,
+		ReceiverConfig:   params.ReceiverConfig,
+		SubscriberConfig: params.SubscriberConfig,
+		Telemetry:        params.Telemetry,
+		Logger:           &params.Logger,
+	})
+
 	if params.TrackInfo.Muted {
 		t.SetMuted(true)
 	}
@@ -139,6 +139,14 @@ func (t *MediaTrack) Source() livekit.TrackSource {
 	return t.params.TrackInfo.Source
 }
 
+func (t *MediaTrack) PublisherID() livekit.ParticipantID {
+	return t.params.ParticipantID
+}
+
+func (t *MediaTrack) PublisherIdentity() livekit.ParticipantIdentity {
+	return t.params.ParticipantIdentity
+}
+
 func (t *MediaTrack) IsSimulcast() bool {
 	return t.simulcasted.Get()
 }
@@ -160,18 +168,7 @@ func (t *MediaTrack) SetMuted(muted bool) {
 	}
 	t.lock.RUnlock()
 
-	// mute all subscribed tracks
-	t.subscribedTracks.Range(func(_, value interface{}) bool {
-		if st, ok := value.(types.SubscribedTrack); ok {
-			st.SetPublisherMuted(muted)
-		}
-		return true
-	})
-
-	// update quality based on subscription if unmuting
-	if !muted {
-		t.updateQualityChange()
-	}
+	t.MediaTrackSubscriptions.SetMuted(muted)
 }
 
 func (t *MediaTrack) AddOnClose(f func()) {
@@ -181,11 +178,6 @@ func (t *MediaTrack) AddOnClose(f func()) {
 	t.onClose = append(t.onClose, f)
 }
 
-func (t *MediaTrack) IsSubscriber(subID livekit.ParticipantID) bool {
-	_, ok := t.subscribedTracks.Load(subID)
-	return ok
-}
-
 func (t *MediaTrack) PublishLossPercentage() uint32 {
 	return FixedPointToPercent(uint8(atomic.LoadUint32(&t.currentUpFracLost)))
 }
@@ -195,19 +187,11 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 	t.lock.Lock()
 	defer t.lock.Unlock()
 
-	subscriberID := sub.ID()
-
-	// don't subscribe to the same track multiple times
-	if _, ok := t.subscribedTracks.Load(subscriberID); ok {
-		return nil
-	}
-
 	if t.receiver == nil {
 		// cannot add, no receiver
 		return errors.New("cannot subscribe without a receiver in place")
 	}
 
-	codec := t.receiver.Codec()
 	// using DownTrack from ion-sfu
 	streamId := string(t.params.ParticipantID)
 	if sub.ProtocolVersion().SupportsPackedStreamId() {
@@ -216,163 +200,27 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 		streamId = PackStreamID(t.params.ParticipantID, t.ID())
 	}
 
-	receiver := NewWrappedReceiver(t.receiver, t.ID(), streamId)
-
-	var rtcpFeedback []webrtc.RTCPFeedback
-	switch t.Kind() {
-	case livekit.TrackType_AUDIO:
-		rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Audio
-	case livekit.TrackType_VIDEO:
-		rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Video
-	}
-	downTrack, err := sfu.NewDownTrack(webrtc.RTPCodecCapability{
-		MimeType:     codec.MimeType,
-		ClockRate:    codec.ClockRate,
-		Channels:     codec.Channels,
-		SDPFmtpLine:  codec.SDPFmtpLine,
-		RTCPFeedback: rtcpFeedback,
-	}, receiver, t.params.BufferFactory, subscriberID, t.params.ReceiverConfig.PacketBufferSize)
+	downTrack, err := t.MediaTrackSubscriptions.AddSubscriber(sub, t.receiver.Codec(), NewWrappedReceiver(t.receiver, t.ID(), streamId))
 	if err != nil {
 		return err
 	}
-	subTrack := NewSubscribedTrack(SubscribedTrackParams{
-		PublisherID:       t.params.ParticipantID,
-		PublisherIdentity: t.params.ParticipantIdentity,
-		SubscriberID:      subscriberID,
-		MediaTrack:        t,
-		DownTrack:         downTrack,
-	})
 
-	var transceiver *webrtc.RTPTransceiver
-	var sender *webrtc.RTPSender
-	if sub.ProtocolVersion().SupportsTransceiverReuse() {
-		//
-		// AddTrack will create a new transceiver or re-use an unused one
-		// if the attributes match. This prevents SDP from bloating
-		// because of dormant transceivers building up.
-		//
-		sender, err = sub.SubscriberPC().AddTrack(downTrack)
-		if err != nil {
-			return err
+	if downTrack != nil {
+		if t.Kind() == livekit.TrackType_AUDIO {
+			downTrack.AddReceiverReportListener(t.handleMaxLossFeedback)
 		}
 
-		// as there is no way to get transceiver from sender, search
-		for _, tr := range sub.SubscriberPC().GetTransceivers() {
-			if tr.Sender() == sender {
-				transceiver = tr
-				break
-			}
-		}
-		if transceiver == nil {
-			// cannot add, no transceiver
-			return errors.New("cannot subscribe without a transceiver in place")
-		}
-	} else {
-		transceiver, err = sub.SubscriberPC().AddTransceiverFromTrack(downTrack, webrtc.RTPTransceiverInit{
-			Direction: webrtc.RTPTransceiverDirectionSendonly,
-		})
-		if err != nil {
-			return err
-		}
-
-		sender = transceiver.Sender()
-		if sender == nil {
-			// cannot add, no sender
-			return errors.New("cannot subscribe without a sender in place")
-		}
+		t.receiver.AddDownTrack(downTrack)
 	}
-
-	sendParameters := sender.GetParameters()
-	downTrack.SetRTPHeaderExtensions(sendParameters.HeaderExtensions)
-
-	downTrack.SetTransceiver(transceiver)
-	// when outtrack is bound, start loop to send reports
-	downTrack.OnBind(func() {
-		go subTrack.Bound()
-		go t.sendDownTrackBindingReports(sub)
-	})
-	downTrack.OnPacketSent(func(_ *sfu.DownTrack, size int) {
-		t.params.Telemetry.OnDownstreamPacket(subscriberID, t.ID(), size)
-	})
-	downTrack.OnPaddingSent(func(_ *sfu.DownTrack, size int) {
-		t.params.Telemetry.OnDownstreamPacket(subscriberID, t.ID(), size)
-	})
-	downTrack.OnRTCP(func(pkts []rtcp.Packet) {
-		t.params.Telemetry.HandleRTCP(livekit.StreamType_DOWNSTREAM, subscriberID, t.ID(), pkts)
-	})
-
-	downTrack.OnCloseHandler(func() {
-		go func() {
-			t.subscribedTracks.Delete(subscriberID)
-			t.params.Telemetry.TrackUnsubscribed(context.Background(), subscriberID, t.ToProto())
-
-			// ignore if the subscribing sub is not connected
-			if sub.SubscriberPC().ConnectionState() == webrtc.PeerConnectionStateClosed {
-				return
-			}
-
-			// if the source has been terminated, we'll need to terminate all of the subscribedtracks
-			// however, if the dest sub has disconnected, then we can skip
-			if sender == nil {
-				return
-			}
-			t.params.Logger.Debugw("removing peerconnection track",
-				"track", t.ID(),
-				"subscriber", sub.Identity(),
-				"subscriberID", subscriberID,
-				"kind", t.Kind(),
-			)
-			if err := sub.SubscriberPC().RemoveTrack(sender); err != nil {
-				if err == webrtc.ErrConnectionClosed {
-					// sub closing, can skip removing subscribedtracks
-					return
-				}
-				if _, ok := err.(*rtcerr.InvalidStateError); !ok {
-					// most of these are safe to ignore, since the track state might have already
-					// been set to Inactive
-					t.params.Logger.Debugw("could not remove remoteTrack from forwarder",
-						"error", err,
-						"subscriber", sub.Identity(),
-						"subscriberID", subscriberID,
-					)
-				}
-			}
-
-			t.NotifySubscriberMute(subscriberID)
-			sub.RemoveSubscribedTrack(subTrack)
-			sub.Negotiate()
-		}()
-	})
-	if t.Kind() == livekit.TrackType_AUDIO {
-		downTrack.AddReceiverReportListener(t.handleMaxLossFeedback)
-	}
-
-	t.subscribedTracks.Store(subscriberID, subTrack)
-	subTrack.SetPublisherMuted(t.IsMuted())
-
-	t.receiver.AddDownTrack(downTrack)
-	// since sub will lock, run it in a goroutine to avoid deadlocks
-	go func() {
-		t.NotifySubscriberMaxQuality(subscriberID, livekit.VideoQuality_HIGH) // start with HIGH, let subscription change it later
-		sub.AddSubscribedTrack(subTrack)
-		sub.Negotiate()
-	}()
-
-	t.params.Telemetry.TrackSubscribed(context.Background(), subscriberID, t.ToProto())
 	return nil
 }
 
 func (t *MediaTrack) NumUpTracks() (uint32, uint32) {
 	numExpected := atomic.LoadUint32(&t.numUpTracks)
 
-	t.maxQualityLock.RLock()
-	maxSubscribed := uint32(0)
-	if !t.allSubscribersMuted {
-		maxSubscribed = uint32(SpatialLayerForQuality(t.maxSubscribedQuality) + 1)
-	}
-	t.maxQualityLock.RUnlock()
-	if maxSubscribed < numExpected {
-		numExpected = maxSubscribed
+	numSubscribedLayers := t.numSubscribedLayers()
+	if numSubscribedLayers < numExpected {
+		numExpected = numSubscribedLayers
 	}
 
 	t.lock.RLock()
@@ -399,10 +247,15 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 	buff.OnFeedback(t.handlePublisherFeedback)
 
 	if t.Kind() == livekit.TrackType_AUDIO {
+		t.audioLevelMu.Lock()
 		t.audioLevel = NewAudioLevel(t.params.AudioConfig.ActiveLevel, t.params.AudioConfig.MinPercentile)
 		buff.OnAudioLevel(func(level uint8, duration uint32) {
+			t.audioLevelMu.RLock()
+			defer t.audioLevelMu.RUnlock()
+
 			t.audioLevel.Observe(level, duration)
 		})
+		t.audioLevelMu.Unlock()
 	} else if t.Kind() == livekit.TrackType_VIDEO {
 		if twcc != nil {
 			buff.OnTransportWideCC(func(sn uint16, timeNS int64, marker bool) {
@@ -471,57 +324,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 	})
 }
 
-// RemoveSubscriber removes participant from subscription
-// stop all forwarders to the client
-func (t *MediaTrack) RemoveSubscriber(participantID livekit.ParticipantID) {
-	subTrack := t.getSubscribedTrack(participantID)
-	if subTrack != nil {
-		go subTrack.DownTrack().Close()
-	}
-}
-
-func (t *MediaTrack) RemoveAllSubscribers() {
-	t.params.Logger.Debugw("removing all subscribers", "track", t.ID())
-	t.lock.Lock()
-	defer t.lock.Unlock()
-	t.subscribedTracks.Range(func(_, val interface{}) bool {
-		if subTrack, ok := val.(types.SubscribedTrack); ok {
-			go subTrack.DownTrack().Close()
-		}
-		return true
-	})
-	t.subscribedTracks = sync.Map{}
-}
-
-func (t *MediaTrack) RevokeDisallowedSubscribers(allowedSubscriberIDs []livekit.ParticipantID) []livekit.ParticipantID {
-	t.lock.Lock()
-	defer t.lock.Unlock()
-
-	var revokedSubscriberIDs []livekit.ParticipantID
-	// LK-TODO: large number of subscribers needs to be solved for this loop
-	t.subscribedTracks.Range(func(key interface{}, val interface{}) bool {
-		if subID, ok := key.(livekit.ParticipantID); ok {
-			found := false
-			for _, allowedID := range allowedSubscriberIDs {
-				if subID == allowedID {
-					found = true
-					break
-				}
-			}
-
-			if !found {
-				if subTrack, ok := val.(types.SubscribedTrack); ok {
-					go subTrack.DownTrack().Close()
-					revokedSubscriberIDs = append(revokedSubscriberIDs, subID)
-				}
-			}
-		}
-		return true
-	})
-
-	return revokedSubscriberIDs
-}
-
 func (t *MediaTrack) ToProto() *livekit.TrackInfo {
 	info := t.params.TrackInfo
 	info.Muted = t.IsMuted()
@@ -538,16 +340,23 @@ func (t *MediaTrack) ToProto() *livekit.TrackInfo {
 	return info
 }
 
+func (t *MediaTrack) GetAudioLevel() (level uint8, active bool) {
+	t.audioLevelMu.RLock()
+	defer t.audioLevelMu.RUnlock()
+
+	if t.audioLevel == nil {
+		return silentAudioLevel, false
+	}
+	return t.audioLevel.GetLevel()
+}
+
 func (t *MediaTrack) UpdateVideoLayers(layers []*livekit.VideoLayer) {
 	for _, layer := range layers {
 		t.layerDimensions.Store(layer.Quality, layer)
 	}
-	t.subscribedTracks.Range(func(_, val interface{}) bool {
-		if st, ok := val.(types.SubscribedTrack); ok {
-			st.UpdateVideoLayer()
-		}
-		return true
-	})
+
+	t.MediaTrackSubscriptions.UpdateVideoLayers()
+
 	// TODO: this might need to trigger a participant update for clients to pick up dimension change
 }
 
@@ -596,53 +405,6 @@ func (t *MediaTrack) GetQualityForDimension(width, height uint32) livekit.VideoQ
 	return quality
 }
 
-func (t *MediaTrack) getSubscribedTrack(subscriberID livekit.ParticipantID) types.SubscribedTrack {
-	if val, ok := t.subscribedTracks.Load(subscriberID); ok {
-		if st, ok := val.(types.SubscribedTrack); ok {
-			return st
-		}
-	}
-	return nil
-}
-
-// TODO: send for all downtracks from the source participant
-// https://tools.ietf.org/html/rfc7941
-func (t *MediaTrack) sendDownTrackBindingReports(sub types.Participant) {
-	var sd []rtcp.SourceDescriptionChunk
-
-	subTrack := t.getSubscribedTrack(sub.ID())
-	if subTrack == nil {
-		return
-	}
-
-	chunks := subTrack.DownTrack().CreateSourceDescriptionChunks()
-	if chunks == nil {
-		return
-	}
-	sd = append(sd, chunks...)
-
-	pkts := []rtcp.Packet{
-		&rtcp.SourceDescription{Chunks: sd},
-	}
-
-	go func() {
-		defer RecoverSilent()
-		batch := pkts
-		i := 0
-		for {
-			if err := sub.SubscriberPC().WriteRTCP(batch); err != nil {
-				t.params.Logger.Errorw("could not write RTCP", err)
-				return
-			}
-			if i > 5 {
-				return
-			}
-			i++
-			time.Sleep(20 * time.Millisecond)
-		}
-	}()
-}
-
 func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) {
 	var maxLost uint8
 	var hasReport bool
@@ -714,17 +476,34 @@ func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) {
 
 // handles max loss for audio packets
 func (t *MediaTrack) handleMaxLossFeedback(_ *sfu.DownTrack, report *rtcp.ReceiverReport) {
-	var (
-		shouldUpdate bool
-		maxLost      uint8
-	)
 	t.statsLock.Lock()
 	for _, rr := range report.Reports {
 		if t.maxDownFracLost < rr.FractionLost {
 			t.maxDownFracLost = rr.FractionLost
 		}
 	}
+	t.statsLock.Unlock()
 
+	t.maybeUpdateLoss()
+}
+
+func (t *MediaTrack) NotifySubscriberNodeMediaLoss(_nodeID string, fractionalLoss uint8) {
+	t.statsLock.Lock()
+	if t.maxDownFracLost < fractionalLoss {
+		t.maxDownFracLost = fractionalLoss
+	}
+	t.statsLock.Unlock()
+
+	t.maybeUpdateLoss()
+}
+
+func (t *MediaTrack) maybeUpdateLoss() {
+	var (
+		shouldUpdate bool
+		maxLost      uint8
+	)
+
+	t.statsLock.Lock()
 	now := time.Now()
 	if now.Sub(t.maxDownFracLostTs) > lostUpdateDelta {
 		shouldUpdate = true
@@ -748,24 +527,16 @@ func (t *MediaTrack) DebugInfo() map[string]interface{} {
 		"PubMuted": t.muted.Get(),
 	}
 
-	subscribedTrackInfo := make([]map[string]interface{}, 0)
-	t.subscribedTracks.Range(func(_, val interface{}) bool {
-		if track, ok := val.(*SubscribedTrack); ok {
-			dt := track.DownTrack().DebugInfo()
-			dt["PubMuted"] = track.pubMuted.Get()
-			dt["SubMuted"] = track.subMuted.Get()
-			subscribedTrackInfo = append(subscribedTrackInfo, dt)
-		}
-		return true
-	})
-	info["DownTracks"] = subscribedTrackInfo
+	info["DownTracks"] = t.MediaTrackSubscriptions.DebugInfo()
 
+	t.lock.RLock()
 	if t.receiver != nil {
 		receiverInfo := t.receiver.DebugInfo()
 		for k, v := range receiverInfo {
 			info[k] = v
 		}
 	}
+	t.lock.RUnlock()
 
 	return info
 }
@@ -816,132 +587,17 @@ func (t *MediaTrack) calculateVideoScore() {
 }
 
 func (t *MediaTrack) OnSubscribedMaxQualityChange(f func(trackID livekit.TrackID, subscribedQualities []*livekit.SubscribedQuality) error) {
-	t.onSubscribedMaxQualityChange = f
-}
-
-func (t *MediaTrack) NotifySubscriberMute(subscriberID livekit.ParticipantID) {
-	if t.Kind() != livekit.TrackType_VIDEO {
-		return
-	}
-
-	t.maxQualityLock.Lock()
-	_, ok := t.maxSubscriberQuality[subscriberID]
-	if !ok {
-		t.maxQualityLock.Unlock()
-		return
-	}
-
-	delete(t.maxSubscriberQuality, subscriberID)
-	t.maxQualityLock.Unlock()
-
-	t.updateQualityChange()
-}
-
-func (t *MediaTrack) NotifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) {
-	if t.Kind() != livekit.TrackType_VIDEO {
-		return
-	}
-
-	t.maxQualityLock.Lock()
-	maxQuality, ok := t.maxSubscriberQuality[subscriberID]
-	if ok && maxQuality == quality {
-		t.maxQualityLock.Unlock()
-		return
-	}
-
-	t.maxSubscriberQuality[subscriberID] = quality
-	t.maxQualityLock.Unlock()
-
-	t.updateQualityChange()
-}
-
-func (t *MediaTrack) startMaxQualityTimer() {
-	t.maxQualityLock.Lock()
-	defer t.maxQualityLock.Unlock()
-
-	if t.Kind() != livekit.TrackType_VIDEO {
-		return
-	}
-
-	t.maxQualityTimer = time.AfterFunc(initialQualityUpdateWait, func() {
-		t.stopMaxQualityTimer()
-		t.updateQualityChange()
-	})
-}
-
-func (t *MediaTrack) stopMaxQualityTimer() {
-	t.maxQualityLock.Lock()
-	defer t.maxQualityLock.Unlock()
-
-	if t.maxQualityTimer != nil {
-		t.maxQualityTimer.Stop()
-		t.maxQualityTimer = nil
-	}
-}
-
-func (t *MediaTrack) updateQualityChange() {
-	if t.Kind() != livekit.TrackType_VIDEO || t.IsMuted() {
-		return
-	}
-
-	var subscribedQualities []*livekit.SubscribedQuality
-
-	t.maxQualityLock.Lock()
-	allSubscribersMuted := false
-	maxSubscribedQuality := livekit.VideoQuality_LOW
-	if len(t.maxSubscriberQuality) == 0 {
-		allSubscribersMuted = true
-	} else {
-		for _, subQuality := range t.maxSubscriberQuality {
-			if subQuality > maxSubscribedQuality {
-				maxSubscribedQuality = subQuality
-			}
+	t.MediaTrackSubscriptions.OnSubscribedMaxQualityChange(func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) {
+		if f != nil && !t.IsMuted() {
+			_ = f(t.ID(), subscribedQualities)
 		}
-	}
 
-	notifyMaxExpected := false
-	maxExpectedSpatialLayer := int32(-1)
-	if allSubscribersMuted {
-		if !t.allSubscribersMuted {
-			notifyMaxExpected = true
-			maxExpectedSpatialLayer = sfu.InvalidLayerSpatial
-
-			t.allSubscribersMuted = true
-
-			subscribedQualities = []*livekit.SubscribedQuality{
-				{Quality: livekit.VideoQuality_LOW, Enabled: false},
-				{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
-				{Quality: livekit.VideoQuality_HIGH, Enabled: false},
-			}
-		}
-	} else {
-		if t.allSubscribersMuted || maxSubscribedQuality != t.maxSubscribedQuality {
-			t.allSubscribersMuted = false
-			notifyMaxExpected = true
-			maxExpectedSpatialLayer = SpatialLayerForQuality(maxSubscribedQuality)
-			t.maxSubscribedQuality = maxSubscribedQuality
-
-			for q := livekit.VideoQuality_LOW; q <= livekit.VideoQuality_HIGH; q++ {
-				subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{
-					Quality: q,
-					Enabled: q <= t.maxSubscribedQuality,
-				})
-			}
-		}
-	}
-	t.maxQualityLock.Unlock()
-
-	if notifyMaxExpected {
 		t.lock.RLock()
 		if t.receiver != nil {
-			t.receiver.SetMaxExpectedSpatialLayer(maxExpectedSpatialLayer)
+			t.receiver.SetMaxExpectedSpatialLayer(SpatialLayerForQuality(maxSubscribedQuality))
 		}
 		t.lock.RUnlock()
-	}
-
-	if len(subscribedQualities) != 0 && t.onSubscribedMaxQualityChange != nil {
-		_ = t.onSubscribedMaxQualityChange(t.ID(), subscribedQualities)
-	}
+	})
 }
 
 //---------------------------
@@ -952,7 +608,11 @@ func SpatialLayerForQuality(quality livekit.VideoQuality) int32 {
 		return 0
 	case livekit.VideoQuality_MEDIUM:
 		return 1
-	default:
+	case livekit.VideoQuality_HIGH:
 		return 2
+	case livekit.VideoQuality_OFF:
+		return -1
+	default:
+		return -1
 	}
 }

pkg/rtc/mediatrack_test.go

@@ -137,7 +137,7 @@ func TestSubscribedMaxQuality(t *testing.T) {
 		})
 
 		// mute all subscribers
-		mt.NotifySubscriberMute("s1")
+		mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF)
 
 		expectedSubscribedQualities := []*livekit.SubscribedQuality{
 			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: false},
@@ -216,7 +216,7 @@ func TestSubscribedMaxQuality(t *testing.T) {
 		require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
 
 		// muting "s2" only should not disable all qualities
-		mt.NotifySubscriberMute("s2")
+		mt.NotifySubscriberMaxQuality("s2", livekit.VideoQuality_OFF)
 
 		expectedSubscribedQualities = []*livekit.SubscribedQuality{
 			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: true},
@@ -227,7 +227,7 @@ func TestSubscribedMaxQuality(t *testing.T) {
 		require.EqualValues(t, expectedSubscribedQualities, actualSubscribedQualities)
 
 		// muting "s1" also should disable all qualities
-		mt.NotifySubscriberMute("s1")
+		mt.NotifySubscriberMaxQuality("s1", livekit.VideoQuality_OFF)
 
 		expectedSubscribedQualities = []*livekit.SubscribedQuality{
 			&livekit.SubscribedQuality{Quality: livekit.VideoQuality_LOW, Enabled: false},

pkg/rtc/mediatracksubscriptions.go

@@ -0,0 +1,543 @@
+package rtc
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
+	"github.com/pion/rtcp"
+	"github.com/pion/webrtc/v3"
+	"github.com/pion/webrtc/v3/pkg/rtcerr"
+
+	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/livekit-server/pkg/sfu"
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+	"github.com/livekit/livekit-server/pkg/telemetry"
+)
+
+const (
+	initialQualityUpdateWait = 10 * time.Second
+)
+
+// MediaTrackSubscriptions manages subscriptions of a media track
+type MediaTrackSubscriptions struct {
+	params MediaTrackSubscriptionsParams
+
+	subscribedTracksMu sync.RWMutex
+	subscribedTracks   map[livekit.ParticipantID]types.SubscribedTrack // participantID => types.SubscribedTrack
+
+	onNoSubscribers func()
+
+	// quality level enable/disable
+	maxQualityLock               sync.RWMutex
+	maxSubscriberQuality         map[livekit.ParticipantID]livekit.VideoQuality
+	maxSubscriberNodeQuality     map[string]livekit.VideoQuality // nodeID => livekit.VideoQuality
+	maxSubscribedQuality         livekit.VideoQuality
+	onSubscribedMaxQualityChange func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality)
+	maxQualityTimer              *time.Timer
+}
+
+type MediaTrackSubscriptionsParams struct {
+	MediaTrack types.MediaTrack
+
+	BufferFactory    *buffer.Factory
+	ReceiverConfig   ReceiverConfig
+	SubscriberConfig DirectionConfig
+
+	Telemetry telemetry.TelemetryService
+
+	Logger *logger.Logger
+}
+
+func NewMediaTrackSubscriptions(params MediaTrackSubscriptionsParams) *MediaTrackSubscriptions {
+	t := &MediaTrackSubscriptions{
+		params:                   params,
+		subscribedTracks:         make(map[livekit.ParticipantID]types.SubscribedTrack),
+		maxSubscriberQuality:     make(map[livekit.ParticipantID]livekit.VideoQuality),
+		maxSubscriberNodeQuality: make(map[string]livekit.VideoQuality),
+	}
+
+	return t
+}
+
+func (t *MediaTrackSubscriptions) OnNoSubscribers(f func()) {
+	t.onNoSubscribers = f
+}
+
+func (t *MediaTrackSubscriptions) SetMuted(muted bool) {
+	t.subscribedTracksMu.RLock()
+	subscribedTracks := t.subscribedTracks
+	t.subscribedTracksMu.RUnlock()
+
+	// mute all subscribed tracks
+	for _, st := range subscribedTracks {
+		st.SetPublisherMuted(muted)
+	}
+
+	// update quality based on subscription if unmuting
+	if !muted {
+		t.updateQualityChange()
+	}
+}
+
+func (t *MediaTrackSubscriptions) IsSubscriber(subID livekit.ParticipantID) bool {
+	t.subscribedTracksMu.RLock()
+	defer t.subscribedTracksMu.RUnlock()
+
+	_, ok := t.subscribedTracks[subID]
+	return ok
+}
+
+// AddSubscriber subscribes sub to current mediaTrack
+func (t *MediaTrackSubscriptions) AddSubscriber(sub types.Participant, codec webrtc.RTPCodecCapability, wr WrappedReceiver) (*sfu.DownTrack, error) {
+	subscriberID := sub.ID()
+
+	t.subscribedTracksMu.Lock()
+	defer t.subscribedTracksMu.Unlock()
+
+	// don't subscribe to the same track multiple times
+	if _, ok := t.subscribedTracks[subscriberID]; ok {
+		return nil, nil
+	}
+
+	var rtcpFeedback []webrtc.RTCPFeedback
+	switch t.params.MediaTrack.Kind() {
+	case livekit.TrackType_AUDIO:
+		rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Audio
+	case livekit.TrackType_VIDEO:
+		rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Video
+	}
+	downTrack, err := sfu.NewDownTrack(webrtc.RTPCodecCapability{
+		MimeType:     codec.MimeType,
+		ClockRate:    codec.ClockRate,
+		Channels:     codec.Channels,
+		SDPFmtpLine:  codec.SDPFmtpLine,
+		RTCPFeedback: rtcpFeedback,
+	}, wr, t.params.BufferFactory, subscriberID, t.params.ReceiverConfig.PacketBufferSize)
+	if err != nil {
+		return nil, err
+	}
+	subTrack := NewSubscribedTrack(SubscribedTrackParams{
+		PublisherID:       t.params.MediaTrack.PublisherID(),
+		PublisherIdentity: t.params.MediaTrack.PublisherIdentity(),
+		SubscriberID:      subscriberID,
+		MediaTrack:        t.params.MediaTrack,
+		DownTrack:         downTrack,
+	})
+
+	var transceiver *webrtc.RTPTransceiver
+	var sender *webrtc.RTPSender
+	if sub.ProtocolVersion().SupportsTransceiverReuse() {
+		//
+		// AddTrack will create a new transceiver or re-use an unused one
+		// if the attributes match. This prevents SDP from bloating
+		// because of dormant transceivers building up.
+		//
+		sender, err = sub.SubscriberPC().AddTrack(downTrack)
+		if err != nil {
+			return nil, err
+		}
+
+		// as there is no way to get transceiver from sender, search
+		for _, tr := range sub.SubscriberPC().GetTransceivers() {
+			if tr.Sender() == sender {
+				transceiver = tr
+				break
+			}
+		}
+		if transceiver == nil {
+			// cannot add, no transceiver
+			return nil, errors.New("cannot subscribe without a transceiver in place")
+		}
+	} else {
+		transceiver, err = sub.SubscriberPC().AddTransceiverFromTrack(downTrack, webrtc.RTPTransceiverInit{
+			Direction: webrtc.RTPTransceiverDirectionSendonly,
+		})
+		if err != nil {
+			return nil, err
+		}
+
+		sender = transceiver.Sender()
+		if sender == nil {
+			// cannot add, no sender
+			return nil, errors.New("cannot subscribe without a sender in place")
+		}
+	}
+
+	sendParameters := sender.GetParameters()
+	downTrack.SetRTPHeaderExtensions(sendParameters.HeaderExtensions)
+
+	downTrack.SetTransceiver(transceiver)
+	// when outtrack is bound, start loop to send reports
+	downTrack.OnBind(func() {
+		go subTrack.Bound()
+		go t.sendDownTrackBindingReports(sub)
+	})
+	downTrack.OnPacketSent(func(_ *sfu.DownTrack, size int) {
+		if t.params.Telemetry != nil {
+			t.params.Telemetry.OnDownstreamPacket(subscriberID, t.params.MediaTrack.ID(), size)
+		}
+	})
+	downTrack.OnPaddingSent(func(_ *sfu.DownTrack, size int) {
+		if t.params.Telemetry != nil {
+			t.params.Telemetry.OnDownstreamPacket(subscriberID, t.params.MediaTrack.ID(), size)
+		}
+	})
+	downTrack.OnRTCP(func(pkts []rtcp.Packet) {
+		if t.params.Telemetry != nil {
+			t.params.Telemetry.HandleRTCP(livekit.StreamType_DOWNSTREAM, subscriberID, t.params.MediaTrack.ID(), pkts)
+		}
+	})
+
+	downTrack.OnCloseHandler(func() {
+		go func() {
+			t.subscribedTracksMu.Lock()
+			delete(t.subscribedTracks, subscriberID)
+			t.subscribedTracksMu.Unlock()
+
+			t.maybeNotifyNoSubscribers()
+			if t.params.Telemetry != nil {
+				t.params.Telemetry.TrackUnsubscribed(context.Background(), subscriberID, t.params.MediaTrack.ToProto())
+			}
+
+			// ignore if the subscribing sub is not connected
+			if sub.SubscriberPC().ConnectionState() == webrtc.PeerConnectionStateClosed {
+				return
+			}
+
+			// if the source has been terminated, we'll need to terminate all of the subscribedtracks
+			// however, if the dest sub has disconnected, then we can skip
+			if sender == nil {
+				return
+			}
+			if t.params.Logger != nil {
+				t.params.Logger.Debugw("removing peerconnection track",
+					"track", t.params.MediaTrack.ID(),
+					"subscriber", sub.Identity(),
+					"subscriberID", subscriberID,
+					"kind", t.params.MediaTrack.Kind(),
+				)
+			}
+			if err := sub.SubscriberPC().RemoveTrack(sender); err != nil {
+				if err == webrtc.ErrConnectionClosed {
+					// sub closing, can skip removing subscribedtracks
+					return
+				}
+				if _, ok := err.(*rtcerr.InvalidStateError); !ok {
+					// most of these are safe to ignore, since the track state might have already
+					// been set to Inactive
+					if t.params.Logger != nil {
+						t.params.Logger.Debugw("could not remove remoteTrack from forwarder",
+							"error", err,
+							"subscriber", sub.Identity(),
+							"subscriberID", subscriberID,
+						)
+					}
+				}
+			}
+
+			t.NotifySubscriberMaxQuality(subscriberID, livekit.VideoQuality_OFF)
+			sub.RemoveSubscribedTrack(subTrack)
+			sub.Negotiate()
+		}()
+	})
+
+	t.subscribedTracks[subscriberID] = subTrack
+	subTrack.SetPublisherMuted(t.params.MediaTrack.IsMuted())
+
+	// since sub will lock, run it in a goroutine to avoid deadlocks
+	go func() {
+		t.NotifySubscriberMaxQuality(subscriberID, livekit.VideoQuality_HIGH) // start with HIGH, let subscription change it later
+		sub.AddSubscribedTrack(subTrack)
+		sub.Negotiate()
+	}()
+
+	if t.params.Telemetry != nil {
+		t.params.Telemetry.TrackSubscribed(context.Background(), subscriberID, t.params.MediaTrack.ToProto())
+	}
+	return downTrack, nil
+}
+
+// RemoveSubscriber removes participant from subscription
+// stop all forwarders to the client
+func (t *MediaTrackSubscriptions) RemoveSubscriber(participantID livekit.ParticipantID) {
+	subTrack := t.getSubscribedTrack(participantID)
+	if subTrack != nil {
+		go subTrack.DownTrack().Close()
+	}
+}
+
+func (t *MediaTrackSubscriptions) RemoveAllSubscribers() {
+	if t.params.Logger != nil {
+		t.params.Logger.Debugw("removing all subscribers", "track", t.params.MediaTrack.ID())
+	}
+
+	t.subscribedTracksMu.RLock()
+	subscribedTracks := t.subscribedTracks
+	t.subscribedTracksMu.RUnlock()
+
+	for _, subTrack := range subscribedTracks {
+		go subTrack.DownTrack().Close()
+	}
+}
+
+func (t *MediaTrackSubscriptions) RevokeDisallowedSubscribers(allowedSubscriberIDs []livekit.ParticipantID) []livekit.ParticipantID {
+	var revokedSubscriberIDs []livekit.ParticipantID
+
+	t.subscribedTracksMu.RLock()
+	subscribedTracks := t.subscribedTracks
+	t.subscribedTracksMu.RUnlock()
+
+	// LK-TODO: large number of subscribers needs to be solved for this loop
+	for subID, subTrack := range subscribedTracks {
+		found := false
+		for _, allowedID := range allowedSubscriberIDs {
+			if subID == allowedID {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			go subTrack.DownTrack().Close()
+			revokedSubscriberIDs = append(revokedSubscriberIDs, subID)
+		}
+	}
+
+	return revokedSubscriberIDs
+}
+
+func (t *MediaTrackSubscriptions) UpdateVideoLayers() {
+	t.subscribedTracksMu.RLock()
+	subscribedTracks := t.subscribedTracks
+	t.subscribedTracksMu.RUnlock()
+
+	for _, st := range subscribedTracks {
+		st.UpdateVideoLayer()
+	}
+}
+
+func (t *MediaTrackSubscriptions) getSubscribedTrack(subscriberID livekit.ParticipantID) types.SubscribedTrack {
+	t.subscribedTracksMu.RLock()
+	defer t.subscribedTracksMu.RUnlock()
+
+	return t.subscribedTracks[subscriberID]
+}
+
+// TODO: send for all downtracks from the source participant
+// https://tools.ietf.org/html/rfc7941
+func (t *MediaTrackSubscriptions) sendDownTrackBindingReports(sub types.Participant) {
+	var sd []rtcp.SourceDescriptionChunk
+
+	subTrack := t.getSubscribedTrack(sub.ID())
+	if subTrack == nil {
+		return
+	}
+
+	chunks := subTrack.DownTrack().CreateSourceDescriptionChunks()
+	if chunks == nil {
+		return
+	}
+	sd = append(sd, chunks...)
+
+	pkts := []rtcp.Packet{
+		&rtcp.SourceDescription{Chunks: sd},
+	}
+
+	go func() {
+		defer RecoverSilent()
+		batch := pkts
+		i := 0
+		for {
+			if err := sub.SubscriberPC().WriteRTCP(batch); err != nil {
+				if t.params.Logger != nil {
+					t.params.Logger.Errorw("could not write RTCP", err)
+				}
+				return
+			}
+			if i > 5 {
+				return
+			}
+			i++
+			time.Sleep(20 * time.Millisecond)
+		}
+	}()
+}
+
+func (t *MediaTrackSubscriptions) DebugInfo() []map[string]interface{} {
+	t.subscribedTracksMu.RLock()
+	subscribedTracks := t.subscribedTracks
+	t.subscribedTracksMu.RUnlock()
+
+	subscribedTrackInfo := make([]map[string]interface{}, 0)
+	for _, val := range subscribedTracks {
+		if st, ok := val.(*SubscribedTrack); ok {
+			dt := st.DownTrack().DebugInfo()
+			dt["PubMuted"] = st.pubMuted.Get()
+			dt["SubMuted"] = st.subMuted.Get()
+			subscribedTrackInfo = append(subscribedTrackInfo, dt)
+		}
+	}
+
+	return subscribedTrackInfo
+}
+
+func (t *MediaTrackSubscriptions) OnSubscribedMaxQualityChange(f func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality)) {
+	t.onSubscribedMaxQualityChange = f
+}
+
+func (t *MediaTrackSubscriptions) NotifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality) {
+	if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO {
+		return
+	}
+
+	t.maxQualityLock.Lock()
+	if quality == livekit.VideoQuality_OFF {
+		_, ok := t.maxSubscriberQuality[subscriberID]
+		if !ok {
+			t.maxQualityLock.Unlock()
+			return
+		}
+
+		delete(t.maxSubscriberQuality, subscriberID)
+	} else {
+		maxQuality, ok := t.maxSubscriberQuality[subscriberID]
+		if ok && maxQuality == quality {
+			t.maxQualityLock.Unlock()
+			return
+		}
+
+		t.maxSubscriberQuality[subscriberID] = quality
+	}
+	t.maxQualityLock.Unlock()
+
+	t.updateQualityChange()
+}
+
+func (t *MediaTrackSubscriptions) NotifySubscriberNodeMaxQuality(nodeID string, quality livekit.VideoQuality) {
+	if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO {
+		return
+	}
+
+	t.maxQualityLock.Lock()
+	if quality == livekit.VideoQuality_OFF {
+		_, ok := t.maxSubscriberNodeQuality[nodeID]
+		if !ok {
+			t.maxQualityLock.Unlock()
+			return
+		}
+
+		delete(t.maxSubscriberNodeQuality, nodeID)
+	} else {
+		maxQuality, ok := t.maxSubscriberNodeQuality[nodeID]
+		if ok && maxQuality == quality {
+			t.maxQualityLock.Unlock()
+			return
+		}
+
+		t.maxSubscriberNodeQuality[nodeID] = quality
+	}
+	t.maxQualityLock.Unlock()
+
+	t.updateQualityChange()
+}
+
+func (t *MediaTrackSubscriptions) startMaxQualityTimer() {
+	t.maxQualityLock.Lock()
+	defer t.maxQualityLock.Unlock()
+
+	if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO {
+		return
+	}
+
+	t.maxQualityTimer = time.AfterFunc(initialQualityUpdateWait, func() {
+		t.stopMaxQualityTimer()
+		t.updateQualityChange()
+	})
+}
+
+func (t *MediaTrackSubscriptions) stopMaxQualityTimer() {
+	t.maxQualityLock.Lock()
+	defer t.maxQualityLock.Unlock()
+
+	if t.maxQualityTimer != nil {
+		t.maxQualityTimer.Stop()
+		t.maxQualityTimer = nil
+	}
+}
+
+func (t *MediaTrackSubscriptions) updateQualityChange() {
+	if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO {
+		return
+	}
+
+	t.maxQualityLock.Lock()
+	maxSubscribedQuality := livekit.VideoQuality_OFF
+	for _, subQuality := range t.maxSubscriberQuality {
+		if maxSubscribedQuality == livekit.VideoQuality_OFF || subQuality > maxSubscribedQuality {
+			maxSubscribedQuality = subQuality
+		}
+	}
+
+	for _, subQuality := range t.maxSubscriberNodeQuality {
+		if maxSubscribedQuality == livekit.VideoQuality_OFF || subQuality > maxSubscribedQuality {
+			maxSubscribedQuality = subQuality
+		}
+	}
+
+	if maxSubscribedQuality == t.maxSubscribedQuality {
+		t.maxQualityLock.Unlock()
+		return
+	}
+
+	t.maxSubscribedQuality = maxSubscribedQuality
+
+	var subscribedQualities []*livekit.SubscribedQuality
+	if t.maxSubscribedQuality == livekit.VideoQuality_OFF {
+		subscribedQualities = []*livekit.SubscribedQuality{
+			{Quality: livekit.VideoQuality_LOW, Enabled: false},
+			{Quality: livekit.VideoQuality_MEDIUM, Enabled: false},
+			{Quality: livekit.VideoQuality_HIGH, Enabled: false},
+		}
+	} else {
+		for q := livekit.VideoQuality_LOW; q <= livekit.VideoQuality_HIGH; q++ {
+			subscribedQualities = append(subscribedQualities, &livekit.SubscribedQuality{
+				Quality: q,
+				Enabled: q <= t.maxSubscribedQuality,
+			})
+		}
+	}
+	t.maxQualityLock.Unlock()
+
+	if t.onSubscribedMaxQualityChange != nil {
+		t.onSubscribedMaxQualityChange(subscribedQualities, maxSubscribedQuality)
+	}
+}
+
+func (t *MediaTrackSubscriptions) numSubscribedLayers() uint32 {
+	t.maxQualityLock.RLock()
+	numSubscribedLayers := uint32(0)
+	if t.maxSubscribedQuality != livekit.VideoQuality_OFF {
+		numSubscribedLayers = uint32(SpatialLayerForQuality(t.maxSubscribedQuality) + 1)
+	}
+	t.maxQualityLock.RUnlock()
+
+	return numSubscribedLayers
+}
+
+func (t *MediaTrackSubscriptions) maybeNotifyNoSubscribers() {
+	if t.onNoSubscribers == nil {
+		return
+	}
+
+	t.subscribedTracksMu.RLock()
+	empty := len(t.subscribedTracks) == 0
+	t.subscribedTracksMu.RUnlock()
+
+	if empty {
+		t.onNoSubscribers()
+	}
+}

pkg/rtc/participant.go

@@ -789,6 +789,14 @@ func (p *ParticipantImpl) SubscriptionPermissionUpdate(publisherID livekit.Parti
 	}
 }
 
+func (p *ParticipantImpl) UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error {
+	return p.uptrackManager.UpdateSubscribedQuality(nodeID, trackID, maxQuality)
+}
+
+func (p *ParticipantImpl) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error {
+	return p.uptrackManager.UpdateMediaLoss(nodeID, trackID, fractionalLoss)
+}
+
 func (p *ParticipantImpl) setupUptrackManager() {
 	p.uptrackManager = NewUptrackManager(UptrackManagerParams{
 		Identity:       p.params.Identity,

pkg/rtc/subscribedtrack.go

@@ -99,7 +99,7 @@ func (t *SubscribedTrack) UpdateVideoLayer() {
 		return
 	}
 	if t.subMuted.Get() {
-		t.MediaTrack().NotifySubscriberMute(t.params.SubscriberID)
+		t.MediaTrack().NotifySubscriberMaxQuality(t.params.SubscriberID, livekit.VideoQuality_OFF)
 		return
 	}
 	settings, ok := t.settings.Load().(*livekit.UpdateTrackSettings)

pkg/rtc/types/interfaces.go

@@ -93,6 +93,10 @@ type Participant interface {
 	UpdateSubscriptionPermissions(permissions *livekit.UpdateSubscriptionPermissions, resolver func(participantID livekit.ParticipantID) Participant) error
 	SubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool)
 
+	UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error
+
+	UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error
+
 	DebugInfo() map[string]interface{}
 }
 
@@ -116,6 +120,11 @@ type MediaTrack interface {
 	Source() livekit.TrackSource
 	IsSimulcast() bool
 
+	PublisherID() livekit.ParticipantID
+	PublisherIdentity() livekit.ParticipantIdentity
+
+	ToProto() *livekit.TrackInfo
+
 	// subscribers
 	AddSubscriber(participant Participant) error
 	RemoveSubscriber(participantID livekit.ParticipantID)
@@ -126,8 +135,10 @@ type MediaTrack interface {
 	// returns quality information that's appropriate for width & height
 	GetQualityForDimension(width, height uint32) livekit.VideoQuality
 
-	NotifySubscriberMute(subscriberID livekit.ParticipantID)
 	NotifySubscriberMaxQuality(subscriberID livekit.ParticipantID, quality livekit.VideoQuality)
+	NotifySubscriberNodeMaxQuality(nodeID string, quality livekit.VideoQuality)
+
+	NotifySubscriberNodeMediaLoss(nodeID string, fractionalLoss uint8)
 }
 
 // PublishedTrack is the main interface representing a track published to the room

pkg/rtc/types/typesfakes/fake_media_track.go

@@ -99,10 +99,37 @@ type FakeMediaTrack struct {
 		arg1 livekit.ParticipantID
 		arg2 livekit.VideoQuality
 	}
-	NotifySubscriberMuteStub        func(livekit.ParticipantID)
-	notifySubscriberMuteMutex       sync.RWMutex
-	notifySubscriberMuteArgsForCall []struct {
-		arg1 livekit.ParticipantID
+	NotifySubscriberNodeMaxQualityStub        func(string, livekit.VideoQuality)
+	notifySubscriberNodeMaxQualityMutex       sync.RWMutex
+	notifySubscriberNodeMaxQualityArgsForCall []struct {
+		arg1 string
+		arg2 livekit.VideoQuality
+	}
+	NotifySubscriberNodeMediaLossStub        func(string, uint8)
+	notifySubscriberNodeMediaLossMutex       sync.RWMutex
+	notifySubscriberNodeMediaLossArgsForCall []struct {
+		arg1 string
+		arg2 uint8
+	}
+	PublisherIDStub        func() livekit.ParticipantID
+	publisherIDMutex       sync.RWMutex
+	publisherIDArgsForCall []struct {
+	}
+	publisherIDReturns struct {
+		result1 livekit.ParticipantID
+	}
+	publisherIDReturnsOnCall map[int]struct {
+		result1 livekit.ParticipantID
+	}
+	PublisherIdentityStub        func() livekit.ParticipantIdentity
+	publisherIdentityMutex       sync.RWMutex
+	publisherIdentityArgsForCall []struct {
+	}
+	publisherIdentityReturns struct {
+		result1 livekit.ParticipantIdentity
+	}
+	publisherIdentityReturnsOnCall map[int]struct {
+		result1 livekit.ParticipantIdentity
 	}
 	RemoveAllSubscribersStub        func()
 	removeAllSubscribersMutex       sync.RWMutex
@@ -139,6 +166,16 @@ type FakeMediaTrack struct {
 	sourceReturnsOnCall map[int]struct {
 		result1 livekit.TrackSource
 	}
+	ToProtoStub        func() *livekit.TrackInfo
+	toProtoMutex       sync.RWMutex
+	toProtoArgsForCall []struct {
+	}
+	toProtoReturns struct {
+		result1 *livekit.TrackInfo
+	}
+	toProtoReturnsOnCall map[int]struct {
+		result1 *livekit.TrackInfo
+	}
 	UpdateVideoLayersStub        func([]*livekit.VideoLayer)
 	updateVideoLayersMutex       sync.RWMutex
 	updateVideoLayersArgsForCall []struct {
@@ -630,36 +667,176 @@ func (fake *FakeMediaTrack) NotifySubscriberMaxQualityArgsForCall(i int) (liveki
 	return argsForCall.arg1, argsForCall.arg2
 }
 
-func (fake *FakeMediaTrack) NotifySubscriberMute(arg1 livekit.ParticipantID) {
-	fake.notifySubscriberMuteMutex.Lock()
-	fake.notifySubscriberMuteArgsForCall = append(fake.notifySubscriberMuteArgsForCall, struct {
-		arg1 livekit.ParticipantID
-	}{arg1})
-	stub := fake.NotifySubscriberMuteStub
-	fake.recordInvocation("NotifySubscriberMute", []interface{}{arg1})
-	fake.notifySubscriberMuteMutex.Unlock()
+func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQuality(arg1 string, arg2 livekit.VideoQuality) {
+	fake.notifySubscriberNodeMaxQualityMutex.Lock()
+	fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct {
+		arg1 string
+		arg2 livekit.VideoQuality
+	}{arg1, arg2})
+	stub := fake.NotifySubscriberNodeMaxQualityStub
+	fake.recordInvocation("NotifySubscriberNodeMaxQuality", []interface{}{arg1, arg2})
+	fake.notifySubscriberNodeMaxQualityMutex.Unlock()
 	if stub != nil {
-		fake.NotifySubscriberMuteStub(arg1)
+		fake.NotifySubscriberNodeMaxQualityStub(arg1, arg2)
 	}
 }
 
-func (fake *FakeMediaTrack) NotifySubscriberMuteCallCount() int {
-	fake.notifySubscriberMuteMutex.RLock()
-	defer fake.notifySubscriberMuteMutex.RUnlock()
-	return len(fake.notifySubscriberMuteArgsForCall)
+func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCallCount() int {
+	fake.notifySubscriberNodeMaxQualityMutex.RLock()
+	defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock()
+	return len(fake.notifySubscriberNodeMaxQualityArgsForCall)
 }
 
-func (fake *FakeMediaTrack) NotifySubscriberMuteCalls(stub func(livekit.ParticipantID)) {
-	fake.notifySubscriberMuteMutex.Lock()
-	defer fake.notifySubscriberMuteMutex.Unlock()
-	fake.NotifySubscriberMuteStub = stub
+func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(string, livekit.VideoQuality)) {
+	fake.notifySubscriberNodeMaxQualityMutex.Lock()
+	defer fake.notifySubscriberNodeMaxQualityMutex.Unlock()
+	fake.NotifySubscriberNodeMaxQualityStub = stub
 }
 
-func (fake *FakeMediaTrack) NotifySubscriberMuteArgsForCall(i int) livekit.ParticipantID {
-	fake.notifySubscriberMuteMutex.RLock()
-	defer fake.notifySubscriberMuteMutex.RUnlock()
-	argsForCall := fake.notifySubscriberMuteArgsForCall[i]
-	return argsForCall.arg1
+func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (string, livekit.VideoQuality) {
+	fake.notifySubscriberNodeMaxQualityMutex.RLock()
+	defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock()
+	argsForCall := fake.notifySubscriberNodeMaxQualityArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLoss(arg1 string, arg2 uint8) {
+	fake.notifySubscriberNodeMediaLossMutex.Lock()
+	fake.notifySubscriberNodeMediaLossArgsForCall = append(fake.notifySubscriberNodeMediaLossArgsForCall, struct {
+		arg1 string
+		arg2 uint8
+	}{arg1, arg2})
+	stub := fake.NotifySubscriberNodeMediaLossStub
+	fake.recordInvocation("NotifySubscriberNodeMediaLoss", []interface{}{arg1, arg2})
+	fake.notifySubscriberNodeMediaLossMutex.Unlock()
+	if stub != nil {
+		fake.NotifySubscriberNodeMediaLossStub(arg1, arg2)
+	}
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossCallCount() int {
+	fake.notifySubscriberNodeMediaLossMutex.RLock()
+	defer fake.notifySubscriberNodeMediaLossMutex.RUnlock()
+	return len(fake.notifySubscriberNodeMediaLossArgsForCall)
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossCalls(stub func(string, uint8)) {
+	fake.notifySubscriberNodeMediaLossMutex.Lock()
+	defer fake.notifySubscriberNodeMediaLossMutex.Unlock()
+	fake.NotifySubscriberNodeMediaLossStub = stub
+}
+
+func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) (string, uint8) {
+	fake.notifySubscriberNodeMediaLossMutex.RLock()
+	defer fake.notifySubscriberNodeMediaLossMutex.RUnlock()
+	argsForCall := fake.notifySubscriberNodeMediaLossArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *FakeMediaTrack) PublisherID() livekit.ParticipantID {
+	fake.publisherIDMutex.Lock()
+	ret, specificReturn := fake.publisherIDReturnsOnCall[len(fake.publisherIDArgsForCall)]
+	fake.publisherIDArgsForCall = append(fake.publisherIDArgsForCall, struct {
+	}{})
+	stub := fake.PublisherIDStub
+	fakeReturns := fake.publisherIDReturns
+	fake.recordInvocation("PublisherID", []interface{}{})
+	fake.publisherIDMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeMediaTrack) PublisherIDCallCount() int {
+	fake.publisherIDMutex.RLock()
+	defer fake.publisherIDMutex.RUnlock()
+	return len(fake.publisherIDArgsForCall)
+}
+
+func (fake *FakeMediaTrack) PublisherIDCalls(stub func() livekit.ParticipantID) {
+	fake.publisherIDMutex.Lock()
+	defer fake.publisherIDMutex.Unlock()
+	fake.PublisherIDStub = stub
+}
+
+func (fake *FakeMediaTrack) PublisherIDReturns(result1 livekit.ParticipantID) {
+	fake.publisherIDMutex.Lock()
+	defer fake.publisherIDMutex.Unlock()
+	fake.PublisherIDStub = nil
+	fake.publisherIDReturns = struct {
+		result1 livekit.ParticipantID
+	}{result1}
+}
+
+func (fake *FakeMediaTrack) PublisherIDReturnsOnCall(i int, result1 livekit.ParticipantID) {
+	fake.publisherIDMutex.Lock()
+	defer fake.publisherIDMutex.Unlock()
+	fake.PublisherIDStub = nil
+	if fake.publisherIDReturnsOnCall == nil {
+		fake.publisherIDReturnsOnCall = make(map[int]struct {
+			result1 livekit.ParticipantID
+		})
+	}
+	fake.publisherIDReturnsOnCall[i] = struct {
+		result1 livekit.ParticipantID
+	}{result1}
+}
+
+func (fake *FakeMediaTrack) PublisherIdentity() livekit.ParticipantIdentity {
+	fake.publisherIdentityMutex.Lock()
+	ret, specificReturn := fake.publisherIdentityReturnsOnCall[len(fake.publisherIdentityArgsForCall)]
+	fake.publisherIdentityArgsForCall = append(fake.publisherIdentityArgsForCall, struct {
+	}{})
+	stub := fake.PublisherIdentityStub
+	fakeReturns := fake.publisherIdentityReturns
+	fake.recordInvocation("PublisherIdentity", []interface{}{})
+	fake.publisherIdentityMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeMediaTrack) PublisherIdentityCallCount() int {
+	fake.publisherIdentityMutex.RLock()
+	defer fake.publisherIdentityMutex.RUnlock()
+	return len(fake.publisherIdentityArgsForCall)
+}
+
+func (fake *FakeMediaTrack) PublisherIdentityCalls(stub func() livekit.ParticipantIdentity) {
+	fake.publisherIdentityMutex.Lock()
+	defer fake.publisherIdentityMutex.Unlock()
+	fake.PublisherIdentityStub = stub
+}
+
+func (fake *FakeMediaTrack) PublisherIdentityReturns(result1 livekit.ParticipantIdentity) {
+	fake.publisherIdentityMutex.Lock()
+	defer fake.publisherIdentityMutex.Unlock()
+	fake.PublisherIdentityStub = nil
+	fake.publisherIdentityReturns = struct {
+		result1 livekit.ParticipantIdentity
+	}{result1}
+}
+
+func (fake *FakeMediaTrack) PublisherIdentityReturnsOnCall(i int, result1 livekit.ParticipantIdentity) {
+	fake.publisherIdentityMutex.Lock()
+	defer fake.publisherIdentityMutex.Unlock()
+	fake.PublisherIdentityStub = nil
+	if fake.publisherIdentityReturnsOnCall == nil {
+		fake.publisherIdentityReturnsOnCall = make(map[int]struct {
+			result1 livekit.ParticipantIdentity
+		})
+	}
+	fake.publisherIdentityReturnsOnCall[i] = struct {
+		result1 livekit.ParticipantIdentity
+	}{result1}
 }
 
 func (fake *FakeMediaTrack) RemoveAllSubscribers() {
@@ -869,6 +1046,59 @@ func (fake *FakeMediaTrack) SourceReturnsOnCall(i int, result1 livekit.TrackSour
 	}{result1}
 }
 
+func (fake *FakeMediaTrack) ToProto() *livekit.TrackInfo {
+	fake.toProtoMutex.Lock()
+	ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)]
+	fake.toProtoArgsForCall = append(fake.toProtoArgsForCall, struct {
+	}{})
+	stub := fake.ToProtoStub
+	fakeReturns := fake.toProtoReturns
+	fake.recordInvocation("ToProto", []interface{}{})
+	fake.toProtoMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeMediaTrack) ToProtoCallCount() int {
+	fake.toProtoMutex.RLock()
+	defer fake.toProtoMutex.RUnlock()
+	return len(fake.toProtoArgsForCall)
+}
+
+func (fake *FakeMediaTrack) ToProtoCalls(stub func() *livekit.TrackInfo) {
+	fake.toProtoMutex.Lock()
+	defer fake.toProtoMutex.Unlock()
+	fake.ToProtoStub = stub
+}
+
+func (fake *FakeMediaTrack) ToProtoReturns(result1 *livekit.TrackInfo) {
+	fake.toProtoMutex.Lock()
+	defer fake.toProtoMutex.Unlock()
+	fake.ToProtoStub = nil
+	fake.toProtoReturns = struct {
+		result1 *livekit.TrackInfo
+	}{result1}
+}
+
+func (fake *FakeMediaTrack) ToProtoReturnsOnCall(i int, result1 *livekit.TrackInfo) {
+	fake.toProtoMutex.Lock()
+	defer fake.toProtoMutex.Unlock()
+	fake.ToProtoStub = nil
+	if fake.toProtoReturnsOnCall == nil {
+		fake.toProtoReturnsOnCall = make(map[int]struct {
+			result1 *livekit.TrackInfo
+		})
+	}
+	fake.toProtoReturnsOnCall[i] = struct {
+		result1 *livekit.TrackInfo
+	}{result1}
+}
+
 func (fake *FakeMediaTrack) UpdateVideoLayers(arg1 []*livekit.VideoLayer) {
 	var arg1Copy []*livekit.VideoLayer
 	if arg1 != nil {
@@ -927,8 +1157,14 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} {
 	defer fake.nameMutex.RUnlock()
 	fake.notifySubscriberMaxQualityMutex.RLock()
 	defer fake.notifySubscriberMaxQualityMutex.RUnlock()
-	fake.notifySubscriberMuteMutex.RLock()
-	defer fake.notifySubscriberMuteMutex.RUnlock()
+	fake.notifySubscriberNodeMaxQualityMutex.RLock()
+	defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock()
+	fake.notifySubscriberNodeMediaLossMutex.RLock()
+	defer fake.notifySubscriberNodeMediaLossMutex.RUnlock()
+	fake.publisherIDMutex.RLock()
+	defer fake.publisherIDMutex.RUnlock()
+	fake.publisherIdentityMutex.RLock()
+	defer fake.publisherIdentityMutex.RUnlock()
 	fake.removeAllSubscribersMutex.RLock()
 	defer fake.removeAllSubscribersMutex.RUnlock()
 	fake.removeSubscriberMutex.RLock()
@@ -939,6 +1175,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} {
 	defer fake.setMutedMutex.RUnlock()
 	fake.sourceMutex.RLock()
 	defer fake.sourceMutex.RUnlock()
+	fake.toProtoMutex.RLock()
+	defer fake.toProtoMutex.RUnlock()
 	fake.updateVideoLayersMutex.RLock()
 	defer fake.updateVideoLayersMutex.RUnlock()
 	copiedInvocations := map[string][][]interface{}{}

pkg/rtc/types/typesfakes/fake_participant.go

@@ -494,6 +494,32 @@ type FakeParticipant struct {
 	toProtoReturnsOnCall map[int]struct {
 		result1 *livekit.ParticipantInfo
 	}
+	UpdateMediaLossStub        func(string, livekit.TrackID, uint32) error
+	updateMediaLossMutex       sync.RWMutex
+	updateMediaLossArgsForCall []struct {
+		arg1 string
+		arg2 livekit.TrackID
+		arg3 uint32
+	}
+	updateMediaLossReturns struct {
+		result1 error
+	}
+	updateMediaLossReturnsOnCall map[int]struct {
+		result1 error
+	}
+	UpdateSubscribedQualityStub        func(string, livekit.TrackID, livekit.VideoQuality) error
+	updateSubscribedQualityMutex       sync.RWMutex
+	updateSubscribedQualityArgsForCall []struct {
+		arg1 string
+		arg2 livekit.TrackID
+		arg3 livekit.VideoQuality
+	}
+	updateSubscribedQualityReturns struct {
+		result1 error
+	}
+	updateSubscribedQualityReturnsOnCall map[int]struct {
+		result1 error
+	}
 	UpdateSubscriptionPermissionsStub        func(*livekit.UpdateSubscriptionPermissions, func(participantID livekit.ParticipantID) types.Participant) error
 	updateSubscriptionPermissionsMutex       sync.RWMutex
 	updateSubscriptionPermissionsArgsForCall []struct {
@@ -3142,6 +3168,132 @@ func (fake *FakeParticipant) ToProtoReturnsOnCall(i int, result1 *livekit.Partic
 	}{result1}
 }
 
+func (fake *FakeParticipant) UpdateMediaLoss(arg1 string, arg2 livekit.TrackID, arg3 uint32) error {
+	fake.updateMediaLossMutex.Lock()
+	ret, specificReturn := fake.updateMediaLossReturnsOnCall[len(fake.updateMediaLossArgsForCall)]
+	fake.updateMediaLossArgsForCall = append(fake.updateMediaLossArgsForCall, struct {
+		arg1 string
+		arg2 livekit.TrackID
+		arg3 uint32
+	}{arg1, arg2, arg3})
+	stub := fake.UpdateMediaLossStub
+	fakeReturns := fake.updateMediaLossReturns
+	fake.recordInvocation("UpdateMediaLoss", []interface{}{arg1, arg2, arg3})
+	fake.updateMediaLossMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2, arg3)
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeParticipant) UpdateMediaLossCallCount() int {
+	fake.updateMediaLossMutex.RLock()
+	defer fake.updateMediaLossMutex.RUnlock()
+	return len(fake.updateMediaLossArgsForCall)
+}
+
+func (fake *FakeParticipant) UpdateMediaLossCalls(stub func(string, livekit.TrackID, uint32) error) {
+	fake.updateMediaLossMutex.Lock()
+	defer fake.updateMediaLossMutex.Unlock()
+	fake.UpdateMediaLossStub = stub
+}
+
+func (fake *FakeParticipant) UpdateMediaLossArgsForCall(i int) (string, livekit.TrackID, uint32) {
+	fake.updateMediaLossMutex.RLock()
+	defer fake.updateMediaLossMutex.RUnlock()
+	argsForCall := fake.updateMediaLossArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
+}
+
+func (fake *FakeParticipant) UpdateMediaLossReturns(result1 error) {
+	fake.updateMediaLossMutex.Lock()
+	defer fake.updateMediaLossMutex.Unlock()
+	fake.UpdateMediaLossStub = nil
+	fake.updateMediaLossReturns = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *FakeParticipant) UpdateMediaLossReturnsOnCall(i int, result1 error) {
+	fake.updateMediaLossMutex.Lock()
+	defer fake.updateMediaLossMutex.Unlock()
+	fake.UpdateMediaLossStub = nil
+	if fake.updateMediaLossReturnsOnCall == nil {
+		fake.updateMediaLossReturnsOnCall = make(map[int]struct {
+			result1 error
+		})
+	}
+	fake.updateMediaLossReturnsOnCall[i] = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *FakeParticipant) UpdateSubscribedQuality(arg1 string, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error {
+	fake.updateSubscribedQualityMutex.Lock()
+	ret, specificReturn := fake.updateSubscribedQualityReturnsOnCall[len(fake.updateSubscribedQualityArgsForCall)]
+	fake.updateSubscribedQualityArgsForCall = append(fake.updateSubscribedQualityArgsForCall, struct {
+		arg1 string
+		arg2 livekit.TrackID
+		arg3 livekit.VideoQuality
+	}{arg1, arg2, arg3})
+	stub := fake.UpdateSubscribedQualityStub
+	fakeReturns := fake.updateSubscribedQualityReturns
+	fake.recordInvocation("UpdateSubscribedQuality", []interface{}{arg1, arg2, arg3})
+	fake.updateSubscribedQualityMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2, arg3)
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeParticipant) UpdateSubscribedQualityCallCount() int {
+	fake.updateSubscribedQualityMutex.RLock()
+	defer fake.updateSubscribedQualityMutex.RUnlock()
+	return len(fake.updateSubscribedQualityArgsForCall)
+}
+
+func (fake *FakeParticipant) UpdateSubscribedQualityCalls(stub func(string, livekit.TrackID, livekit.VideoQuality) error) {
+	fake.updateSubscribedQualityMutex.Lock()
+	defer fake.updateSubscribedQualityMutex.Unlock()
+	fake.UpdateSubscribedQualityStub = stub
+}
+
+func (fake *FakeParticipant) UpdateSubscribedQualityArgsForCall(i int) (string, livekit.TrackID, livekit.VideoQuality) {
+	fake.updateSubscribedQualityMutex.RLock()
+	defer fake.updateSubscribedQualityMutex.RUnlock()
+	argsForCall := fake.updateSubscribedQualityArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
+}
+
+func (fake *FakeParticipant) UpdateSubscribedQualityReturns(result1 error) {
+	fake.updateSubscribedQualityMutex.Lock()
+	defer fake.updateSubscribedQualityMutex.Unlock()
+	fake.UpdateSubscribedQualityStub = nil
+	fake.updateSubscribedQualityReturns = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *FakeParticipant) UpdateSubscribedQualityReturnsOnCall(i int, result1 error) {
+	fake.updateSubscribedQualityMutex.Lock()
+	defer fake.updateSubscribedQualityMutex.Unlock()
+	fake.UpdateSubscribedQualityStub = nil
+	if fake.updateSubscribedQualityReturnsOnCall == nil {
+		fake.updateSubscribedQualityReturnsOnCall = make(map[int]struct {
+			result1 error
+		})
+	}
+	fake.updateSubscribedQualityReturnsOnCall[i] = struct {
+		result1 error
+	}{result1}
+}
+
 func (fake *FakeParticipant) UpdateSubscriptionPermissions(arg1 *livekit.UpdateSubscriptionPermissions, arg2 func(participantID livekit.ParticipantID) types.Participant) error {
 	fake.updateSubscriptionPermissionsMutex.Lock()
 	ret, specificReturn := fake.updateSubscriptionPermissionsReturnsOnCall[len(fake.updateSubscriptionPermissionsArgsForCall)]
@@ -3315,6 +3467,10 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
 	defer fake.subscriptionPermissionUpdateMutex.RUnlock()
 	fake.toProtoMutex.RLock()
 	defer fake.toProtoMutex.RUnlock()
+	fake.updateMediaLossMutex.RLock()
+	defer fake.updateMediaLossMutex.RUnlock()
+	fake.updateSubscribedQualityMutex.RLock()
+	defer fake.updateSubscribedQualityMutex.RUnlock()
 	fake.updateSubscriptionPermissionsMutex.RLock()
 	defer fake.updateSubscriptionPermissionsMutex.RUnlock()
 	copiedInvocations := map[string][][]interface{}{}

pkg/rtc/types/typesfakes/fake_published_track.go

@@ -115,10 +115,17 @@ type FakePublishedTrack struct {
 		arg1 livekit.ParticipantID
 		arg2 livekit.VideoQuality
 	}
-	NotifySubscriberMuteStub        func(livekit.ParticipantID)
-	notifySubscriberMuteMutex       sync.RWMutex
-	notifySubscriberMuteArgsForCall []struct {
-		arg1 livekit.ParticipantID
+	NotifySubscriberNodeMaxQualityStub        func(string, livekit.VideoQuality)
+	notifySubscriberNodeMaxQualityMutex       sync.RWMutex
+	notifySubscriberNodeMaxQualityArgsForCall []struct {
+		arg1 string
+		arg2 livekit.VideoQuality
+	}
+	NotifySubscriberNodeMediaLossStub        func(string, uint8)
+	notifySubscriberNodeMediaLossMutex       sync.RWMutex
+	notifySubscriberNodeMediaLossArgsForCall []struct {
+		arg1 string
+		arg2 uint8
 	}
 	NumUpTracksStub        func() (uint32, uint32)
 	numUpTracksMutex       sync.RWMutex
@@ -142,6 +149,26 @@ type FakePublishedTrack struct {
 	publishLossPercentageReturnsOnCall map[int]struct {
 		result1 uint32
 	}
+	PublisherIDStub        func() livekit.ParticipantID
+	publisherIDMutex       sync.RWMutex
+	publisherIDArgsForCall []struct {
+	}
+	publisherIDReturns struct {
+		result1 livekit.ParticipantID
+	}
+	publisherIDReturnsOnCall map[int]struct {
+		result1 livekit.ParticipantID
+	}
+	PublisherIdentityStub        func() livekit.ParticipantIdentity
+	publisherIdentityMutex       sync.RWMutex
+	publisherIdentityArgsForCall []struct {
+	}
+	publisherIdentityReturns struct {
+		result1 livekit.ParticipantIdentity
+	}
+	publisherIdentityReturnsOnCall map[int]struct {
+		result1 livekit.ParticipantIdentity
+	}
 	ReceiverStub        func() sfu.TrackReceiver
 	receiverMutex       sync.RWMutex
 	receiverArgsForCall []struct {
@@ -793,36 +820,70 @@ func (fake *FakePublishedTrack) NotifySubscriberMaxQualityArgsForCall(i int) (li
 	return argsForCall.arg1, argsForCall.arg2
 }
 
-func (fake *FakePublishedTrack) NotifySubscriberMute(arg1 livekit.ParticipantID) {
-	fake.notifySubscriberMuteMutex.Lock()
-	fake.notifySubscriberMuteArgsForCall = append(fake.notifySubscriberMuteArgsForCall, struct {
-		arg1 livekit.ParticipantID
-	}{arg1})
-	stub := fake.NotifySubscriberMuteStub
-	fake.recordInvocation("NotifySubscriberMute", []interface{}{arg1})
-	fake.notifySubscriberMuteMutex.Unlock()
+func (fake *FakePublishedTrack) NotifySubscriberNodeMaxQuality(arg1 string, arg2 livekit.VideoQuality) {
+	fake.notifySubscriberNodeMaxQualityMutex.Lock()
+	fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct {
+		arg1 string
+		arg2 livekit.VideoQuality
+	}{arg1, arg2})
+	stub := fake.NotifySubscriberNodeMaxQualityStub
+	fake.recordInvocation("NotifySubscriberNodeMaxQuality", []interface{}{arg1, arg2})
+	fake.notifySubscriberNodeMaxQualityMutex.Unlock()
 	if stub != nil {
-		fake.NotifySubscriberMuteStub(arg1)
+		fake.NotifySubscriberNodeMaxQualityStub(arg1, arg2)
 	}
 }
 
-func (fake *FakePublishedTrack) NotifySubscriberMuteCallCount() int {
-	fake.notifySubscriberMuteMutex.RLock()
-	defer fake.notifySubscriberMuteMutex.RUnlock()
-	return len(fake.notifySubscriberMuteArgsForCall)
+func (fake *FakePublishedTrack) NotifySubscriberNodeMaxQualityCallCount() int {
+	fake.notifySubscriberNodeMaxQualityMutex.RLock()
+	defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock()
+	return len(fake.notifySubscriberNodeMaxQualityArgsForCall)
 }
 
-func (fake *FakePublishedTrack) NotifySubscriberMuteCalls(stub func(livekit.ParticipantID)) {
-	fake.notifySubscriberMuteMutex.Lock()
-	defer fake.notifySubscriberMuteMutex.Unlock()
-	fake.NotifySubscriberMuteStub = stub
+func (fake *FakePublishedTrack) NotifySubscriberNodeMaxQualityCalls(stub func(string, livekit.VideoQuality)) {
+	fake.notifySubscriberNodeMaxQualityMutex.Lock()
+	defer fake.notifySubscriberNodeMaxQualityMutex.Unlock()
+	fake.NotifySubscriberNodeMaxQualityStub = stub
 }
 
-func (fake *FakePublishedTrack) NotifySubscriberMuteArgsForCall(i int) livekit.ParticipantID {
-	fake.notifySubscriberMuteMutex.RLock()
-	defer fake.notifySubscriberMuteMutex.RUnlock()
-	argsForCall := fake.notifySubscriberMuteArgsForCall[i]
-	return argsForCall.arg1
+func (fake *FakePublishedTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (string, livekit.VideoQuality) {
+	fake.notifySubscriberNodeMaxQualityMutex.RLock()
+	defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock()
+	argsForCall := fake.notifySubscriberNodeMaxQualityArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberNodeMediaLoss(arg1 string, arg2 uint8) {
+	fake.notifySubscriberNodeMediaLossMutex.Lock()
+	fake.notifySubscriberNodeMediaLossArgsForCall = append(fake.notifySubscriberNodeMediaLossArgsForCall, struct {
+		arg1 string
+		arg2 uint8
+	}{arg1, arg2})
+	stub := fake.NotifySubscriberNodeMediaLossStub
+	fake.recordInvocation("NotifySubscriberNodeMediaLoss", []interface{}{arg1, arg2})
+	fake.notifySubscriberNodeMediaLossMutex.Unlock()
+	if stub != nil {
+		fake.NotifySubscriberNodeMediaLossStub(arg1, arg2)
+	}
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberNodeMediaLossCallCount() int {
+	fake.notifySubscriberNodeMediaLossMutex.RLock()
+	defer fake.notifySubscriberNodeMediaLossMutex.RUnlock()
+	return len(fake.notifySubscriberNodeMediaLossArgsForCall)
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberNodeMediaLossCalls(stub func(string, uint8)) {
+	fake.notifySubscriberNodeMediaLossMutex.Lock()
+	defer fake.notifySubscriberNodeMediaLossMutex.Unlock()
+	fake.NotifySubscriberNodeMediaLossStub = stub
+}
+
+func (fake *FakePublishedTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) (string, uint8) {
+	fake.notifySubscriberNodeMediaLossMutex.RLock()
+	defer fake.notifySubscriberNodeMediaLossMutex.RUnlock()
+	argsForCall := fake.notifySubscriberNodeMediaLossArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
 }
 
 func (fake *FakePublishedTrack) NumUpTracks() (uint32, uint32) {
@@ -934,6 +995,112 @@ func (fake *FakePublishedTrack) PublishLossPercentageReturnsOnCall(i int, result
 	}{result1}
 }
 
+func (fake *FakePublishedTrack) PublisherID() livekit.ParticipantID {
+	fake.publisherIDMutex.Lock()
+	ret, specificReturn := fake.publisherIDReturnsOnCall[len(fake.publisherIDArgsForCall)]
+	fake.publisherIDArgsForCall = append(fake.publisherIDArgsForCall, struct {
+	}{})
+	stub := fake.PublisherIDStub
+	fakeReturns := fake.publisherIDReturns
+	fake.recordInvocation("PublisherID", []interface{}{})
+	fake.publisherIDMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakePublishedTrack) PublisherIDCallCount() int {
+	fake.publisherIDMutex.RLock()
+	defer fake.publisherIDMutex.RUnlock()
+	return len(fake.publisherIDArgsForCall)
+}
+
+func (fake *FakePublishedTrack) PublisherIDCalls(stub func() livekit.ParticipantID) {
+	fake.publisherIDMutex.Lock()
+	defer fake.publisherIDMutex.Unlock()
+	fake.PublisherIDStub = stub
+}
+
+func (fake *FakePublishedTrack) PublisherIDReturns(result1 livekit.ParticipantID) {
+	fake.publisherIDMutex.Lock()
+	defer fake.publisherIDMutex.Unlock()
+	fake.PublisherIDStub = nil
+	fake.publisherIDReturns = struct {
+		result1 livekit.ParticipantID
+	}{result1}
+}
+
+func (fake *FakePublishedTrack) PublisherIDReturnsOnCall(i int, result1 livekit.ParticipantID) {
+	fake.publisherIDMutex.Lock()
+	defer fake.publisherIDMutex.Unlock()
+	fake.PublisherIDStub = nil
+	if fake.publisherIDReturnsOnCall == nil {
+		fake.publisherIDReturnsOnCall = make(map[int]struct {
+			result1 livekit.ParticipantID
+		})
+	}
+	fake.publisherIDReturnsOnCall[i] = struct {
+		result1 livekit.ParticipantID
+	}{result1}
+}
+
+func (fake *FakePublishedTrack) PublisherIdentity() livekit.ParticipantIdentity {
+	fake.publisherIdentityMutex.Lock()
+	ret, specificReturn := fake.publisherIdentityReturnsOnCall[len(fake.publisherIdentityArgsForCall)]
+	fake.publisherIdentityArgsForCall = append(fake.publisherIdentityArgsForCall, struct {
+	}{})
+	stub := fake.PublisherIdentityStub
+	fakeReturns := fake.publisherIdentityReturns
+	fake.recordInvocation("PublisherIdentity", []interface{}{})
+	fake.publisherIdentityMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakePublishedTrack) PublisherIdentityCallCount() int {
+	fake.publisherIdentityMutex.RLock()
+	defer fake.publisherIdentityMutex.RUnlock()
+	return len(fake.publisherIdentityArgsForCall)
+}
+
+func (fake *FakePublishedTrack) PublisherIdentityCalls(stub func() livekit.ParticipantIdentity) {
+	fake.publisherIdentityMutex.Lock()
+	defer fake.publisherIdentityMutex.Unlock()
+	fake.PublisherIdentityStub = stub
+}
+
+func (fake *FakePublishedTrack) PublisherIdentityReturns(result1 livekit.ParticipantIdentity) {
+	fake.publisherIdentityMutex.Lock()
+	defer fake.publisherIdentityMutex.Unlock()
+	fake.PublisherIdentityStub = nil
+	fake.publisherIdentityReturns = struct {
+		result1 livekit.ParticipantIdentity
+	}{result1}
+}
+
+func (fake *FakePublishedTrack) PublisherIdentityReturnsOnCall(i int, result1 livekit.ParticipantIdentity) {
+	fake.publisherIdentityMutex.Lock()
+	defer fake.publisherIdentityMutex.Unlock()
+	fake.PublisherIdentityStub = nil
+	if fake.publisherIdentityReturnsOnCall == nil {
+		fake.publisherIdentityReturnsOnCall = make(map[int]struct {
+			result1 livekit.ParticipantIdentity
+		})
+	}
+	fake.publisherIdentityReturnsOnCall[i] = struct {
+		result1 livekit.ParticipantIdentity
+	}{result1}
+}
+
 func (fake *FakePublishedTrack) Receiver() sfu.TrackReceiver {
 	fake.receiverMutex.Lock()
 	ret, specificReturn := fake.receiverReturnsOnCall[len(fake.receiverArgsForCall)]
@@ -1415,12 +1582,18 @@ func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} {
 	defer fake.nameMutex.RUnlock()
 	fake.notifySubscriberMaxQualityMutex.RLock()
 	defer fake.notifySubscriberMaxQualityMutex.RUnlock()
-	fake.notifySubscriberMuteMutex.RLock()
-	defer fake.notifySubscriberMuteMutex.RUnlock()
+	fake.notifySubscriberNodeMaxQualityMutex.RLock()
+	defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock()
+	fake.notifySubscriberNodeMediaLossMutex.RLock()
+	defer fake.notifySubscriberNodeMediaLossMutex.RUnlock()
 	fake.numUpTracksMutex.RLock()
 	defer fake.numUpTracksMutex.RUnlock()
 	fake.publishLossPercentageMutex.RLock()
 	defer fake.publishLossPercentageMutex.RUnlock()
+	fake.publisherIDMutex.RLock()
+	defer fake.publisherIDMutex.RUnlock()
+	fake.publisherIdentityMutex.RLock()
+	defer fake.publisherIdentityMutex.RUnlock()
 	fake.receiverMutex.RLock()
 	defer fake.receiverMutex.RUnlock()
 	fake.removeAllSubscribersMutex.RLock()

pkg/rtc/uptrackmanager.go

@@ -1,6 +1,7 @@
 package rtc
 
 import (
+	"errors"
 	"sync"
 
 	"github.com/livekit/protocol/livekit"
@@ -232,10 +233,7 @@ func (u *UptrackManager) GetAudioLevel() (level uint8, active bool) {
 	level = silentAudioLevel
 	for _, pt := range u.publishedTracks {
 		if mt, ok := pt.(*MediaTrack); ok {
-			if mt.audioLevel == nil {
-				continue
-			}
-			tl, ta := mt.audioLevel.GetLevel()
+			tl, ta := mt.GetAudioLevel()
 			if ta {
 				active = true
 				if tl < level {
@@ -314,6 +312,40 @@ func (u *UptrackManager) UpdateSubscriptionPermissions(
 	return nil
 }
 
+func (u *UptrackManager) UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error {
+	u.lock.RLock()
+	defer u.lock.RUnlock()
+
+	track := u.getPublishedTrack(trackID)
+	if track == nil {
+		u.params.Logger.Warnw("could not find track", nil, "trackID", trackID)
+		return errors.New("could not find track")
+	}
+
+	if mt, ok := track.(*MediaTrack); ok {
+		mt.NotifySubscriberNodeMaxQuality(nodeID, maxQuality)
+	}
+
+	return nil
+}
+
+func (u *UptrackManager) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error {
+	u.lock.RLock()
+	defer u.lock.RUnlock()
+
+	track := u.getPublishedTrack(trackID)
+	if track == nil {
+		u.params.Logger.Warnw("could not find track", nil, "trackID", trackID)
+		return errors.New("could not find track")
+	}
+
+	if mt, ok := track.(*MediaTrack); ok {
+		mt.NotifySubscriberNodeMediaLoss(nodeID, uint8(fractionalLoss))
+	}
+
+	return nil
+}
+
 // when a new remoteTrack is created, creates a Track and adds it to room
 func (u *UptrackManager) MediaTrackReceived(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
 	var newTrack bool