Fixed unclean DownTrack close when removed before bound. (#736)

* Fixed unclean DownTrack close when removed before bound.

When a DownTrack is closed before it had a chance to be bound to a
transceiver, we'd skip close and leave it hanging. This is unlikely in
normal operations. However, it can be seen with permissions and
subscription APIs.

* remove remaining peerID references

pkg/rtc/mediatrack.go

@@ -154,7 +154,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 		newWR := sfu.NewWebRTCReceiver(
 			receiver,
 			track,
-			t.PublisherID(),
 			t.params.TrackInfo,
 			LoggerWithCodecMime(t.params.Logger, mime),
 			twcc,

pkg/rtc/mediatracksubscriptions.go

@@ -144,6 +144,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *
 		codecs,
 		wr,
 		t.params.BufferFactory,
+		sub.Identity(),
 		subscriberID,
 		t.params.ReceiverConfig.PacketBufferSize,
 		LoggerWithTrack(sub.GetLogger(), trackID),
@@ -309,6 +310,10 @@ func (t *MediaTrackSubscriptions) RevokeDisallowedSubscribers(allowedSubscriberI
 		}
 
 		if !found {
+			t.params.Logger.Infow("revoking subscription",
+				"subscriber", subTrack.SubscriberIdentity(),
+				"subscriberID", subTrack.SubscriberID(),
+			)
 			go subTrack.DownTrack().Close()
 			revokedSubscriberIdentities = append(revokedSubscriberIdentities, subTrack.SubscriberIdentity())
 		}

pkg/rtc/signalhandler.go

@@ -38,7 +38,8 @@ func HandleParticipantSignal(room types.Room, participant types.LocalParticipant
 		participant.SetTrackMuted(livekit.TrackID(msg.Mute.Sid), msg.Mute.Muted, false)
 	case *livekit.SignalRequest_Subscription:
 		var err error
-		if participant.CanSubscribe() {
+		// always allow unsubscribe
+		if participant.CanSubscribe() || !msg.Subscription.Subscribe {
 			updateErr := room.UpdateSubscriptions(
 				participant,
 				livekit.StringsAsTrackIDs(msg.Subscription.TrackSids),

pkg/rtc/uptrackmanager.go

@@ -198,21 +198,21 @@ func (u *UpTrackManager) UpdateSubscriptionPermission(
 	u.lock.Lock()
 	defer u.lock.Unlock()
 
-	// store as is for use when migrating
-	u.subscriptionPermission = subscriptionPermission
 	if subscriptionPermission == nil {
+		// store as is for use when migrating
+		u.subscriptionPermission = subscriptionPermission
 		// possible to get a nil when migrating
 		return nil
 	}
 
 	if err := u.parseSubscriptionPermissions(subscriptionPermission, resolverBySid); err != nil {
-		// do not accept permissions if parse fails
-		u.subscriptionPermission = nil
+		// when failed, do not override previous permissions
 		return err
 	}
 
+	// store as is for use when migrating
+	u.subscriptionPermission = subscriptionPermission
 	u.processPendingSubscriptions(resolverByIdentity)
-
 	u.maybeRevokeSubscriptions(resolverByIdentity)
 
 	return nil
@@ -322,19 +322,15 @@ func (u *UpTrackManager) parseSubscriptionPermissions(
 	}
 
 	// per participant permissions
-	u.subscriberPermissions = make(map[livekit.ParticipantIdentity]*livekit.TrackPermission)
+	subscriberPermissions := make(map[livekit.ParticipantIdentity]*livekit.TrackPermission)
 	for _, trackPerms := range subscriptionPermission.TrackPermissions {
 		subscriberIdentity := livekit.ParticipantIdentity(trackPerms.ParticipantIdentity)
 		if subscriberIdentity == "" {
 			if trackPerms.ParticipantSid == "" {
-				u.subscriberPermissions = nil
 				return ErrSubscriptionPermissionNeedsId
 			}
 
-			var sub types.LocalParticipant
-			if resolver != nil {
-				sub = resolver(livekit.ParticipantID(trackPerms.ParticipantSid))
-			}
+			sub := resolver(livekit.ParticipantID(trackPerms.ParticipantSid))
 			if sub == nil {
 				u.params.Logger.Warnw("could not find subscriber for permissions update", nil, "subscriberID", trackPerms.ParticipantSid)
 				continue
@@ -353,9 +349,11 @@ func (u *UpTrackManager) parseSubscriptionPermissions(
 			}
 		}
 
-		u.subscriberPermissions[subscriberIdentity] = trackPerms
+		subscriberPermissions[subscriberIdentity] = trackPerms
 	}
 
+	u.subscriberPermissions = subscriberPermissions
+
 	return nil
 }
 
@@ -437,11 +435,13 @@ func (u *UpTrackManager) maybeRemovePendingSubscription(trackID livekit.TrackID,
 	}
 }
 
+// creates subscriptions for tracks if permissions have been granted
 func (u *UpTrackManager) processPendingSubscriptions(resolver func(participantIdentity livekit.ParticipantIdentity) types.LocalParticipant) {
 	updatedPendingSubscriptions := make(map[livekit.TrackID][]livekit.ParticipantIdentity)
 	for trackID, pending := range u.pendingSubscriptions {
 		track := u.getPublishedTrack(trackID)
 		if track == nil {
+			// published track is gone
 			continue
 		}
 
@@ -488,11 +488,9 @@ func (u *UpTrackManager) maybeRevokeSubscriptions(resolver func(participantIdent
 
 		revokedSubscribers := track.RevokeDisallowedSubscribers(allowed)
 		for _, subIdentity := range revokedSubscribers {
-			var sub types.LocalParticipant
-			if resolver != nil {
-				sub = resolver(subIdentity)
-			}
+			sub := resolver(subIdentity)
 			if sub == nil {
+				// participant may have disconnected
 				continue
 			}
 

pkg/rtc/utils.go

@@ -35,12 +35,12 @@ func PackDataTrackLabel(participantID livekit.ParticipantID, trackID livekit.Tra
 	return string(participantID) + trackIdSeparator + string(trackID) + trackIdSeparator + label
 }
 
-func UnpackDataTrackLabel(packed string) (peerID livekit.ParticipantID, trackID livekit.TrackID, label string) {
+func UnpackDataTrackLabel(packed string) (participantID livekit.ParticipantID, trackID livekit.TrackID, label string) {
 	parts := strings.Split(packed, trackIdSeparator)
 	if len(parts) != 3 {
 		return "", livekit.TrackID(packed), ""
 	}
-	peerID = livekit.ParticipantID(parts[0])
+	participantID = livekit.ParticipantID(parts[0])
 	trackID = livekit.TrackID(parts[1])
 	label = parts[2]
 	return

pkg/sfu/downtrack.go

@@ -3,7 +3,6 @@ package sfu
 import (
 	"encoding/binary"
 	"errors"
-	"fmt"
 	"io"
 	"strings"
 	"sync"
@@ -30,9 +29,11 @@ type TrackSender interface {
 	UpTrackBitrateAvailabilityChange()
 	WriteRTP(p *buffer.ExtPacket, layer int32) error
 	Close()
+	IsClosed() bool
 	// ID is the globally unique identifier for this Track.
 	ID() string
-	PeerID() livekit.ParticipantID
+	SubscriberID() livekit.ParticipantID
+	SubscriberIdentity() livekit.ParticipantIdentity
 }
 
 const (
@@ -57,7 +58,8 @@ var (
 	ErrNotVP8                            = errors.New("not VP8")
 	ErrOutOfOrderVP8PictureIdCacheMiss   = errors.New("out-of-order VP8 picture id not found in cache")
 	ErrFilteredVP8TemporalLayer          = errors.New("filtered VP8 temporal layer")
-	ErrTrackAlreadyBind                  = errors.New("already bind")
+	ErrDownTrackAlreadyBound             = errors.New("already bound")
+	ErrDownTrackClosed                   = errors.New("downtrack closed")
 )
 
 var (
@@ -101,19 +103,25 @@ type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport)
 // DownTrack implements TrackLocal, is the track used to write packets
 // to SFU Subscriber, the track handle the packets for simple, simulcast
 // and SVC Publisher.
+// A DownTrack has the following lifecycle
+// - new
+// - bound / unbound
+// - closed
+// once closed, a DownTrack cannot be re-used.
 type DownTrack struct {
-	logger        logger.Logger
-	id            livekit.TrackID
-	peerID        livekit.ParticipantID
-	bound         atomic.Bool
-	kind          webrtc.RTPCodecType
-	mime          string
-	ssrc          uint32
-	streamID      string
-	maxTrack      int
-	payloadType   uint8
-	sequencer     *sequencer
-	bufferFactory *buffer.Factory
+	logger             logger.Logger
+	id                 livekit.TrackID
+	subscriberIdentity livekit.ParticipantIdentity
+	subscriberID       livekit.ParticipantID
+	bound              atomic.Bool
+	kind               webrtc.RTPCodecType
+	mime               string
+	ssrc               uint32
+	streamID           string
+	maxTrack           int
+	payloadType        uint8
+	sequencer          *sequencer
+	bufferFactory      *buffer.Factory
 
 	forwarder *Forwarder
 
@@ -129,7 +137,7 @@ type DownTrack struct {
 	onBind                  func()
 	receiverReportListeners []ReceiverReportListener
 	listenerLock            sync.RWMutex
-	closeOnce               sync.Once
+	isClosed                atomic.Bool
 
 	rtpStats *buffer.RTPStats
 
@@ -186,7 +194,8 @@ func NewDownTrack(
 	codecs []webrtc.RTPCodecCapability,
 	r TrackReceiver,
 	bf *buffer.Factory,
-	peerID livekit.ParticipantID,
+	subIdentity livekit.ParticipantIdentity,
+	subID livekit.ParticipantID,
 	mt int,
 	logger logger.Logger,
 ) (*DownTrack, error) {
@@ -201,15 +210,16 @@ func NewDownTrack(
 	}
 
 	d := &DownTrack{
-		logger:         logger,
-		id:             r.TrackID(),
-		peerID:         peerID,
-		maxTrack:       mt,
-		streamID:       r.StreamID(),
-		bufferFactory:  bf,
-		receiver:       r,
-		upstreamCodecs: codecs,
-		kind:           kind,
+		logger:             logger,
+		id:                 r.TrackID(),
+		subscriberIdentity: subIdentity,
+		subscriberID:       subID,
+		maxTrack:           mt,
+		streamID:           r.StreamID(),
+		bufferFactory:      bf,
+		receiver:           r,
+		upstreamCodecs:     codecs,
+		kind:               kind,
 	}
 	d.forwarder = NewForwarder(d.kind, d.logger)
 
@@ -255,8 +265,11 @@ func NewDownTrack(
 // This asserts that the code requested is supported by the remote peer.
 // If so it sets up all the state (SSRC and PayloadType) to have a call
 func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, error) {
+	if d.IsClosed() {
+		return webrtc.RTPCodecParameters{}, ErrDownTrackClosed
+	}
 	if d.bound.Load() {
-		return webrtc.RTPCodecParameters{}, ErrTrackAlreadyBind
+		return webrtc.RTPCodecParameters{}, ErrDownTrackAlreadyBound
 	}
 	var codec webrtc.RTPCodecParameters
 	for _, c := range d.upstreamCodecs {
@@ -316,7 +329,9 @@ func (d *DownTrack) Codec() webrtc.RTPCodecCapability { return d.codec }
 // StreamID is the group this track belongs too. This must be unique
 func (d *DownTrack) StreamID() string { return d.streamID }
 
-func (d *DownTrack) PeerID() livekit.ParticipantID { return d.peerID }
+func (d *DownTrack) SubscriberIdentity() livekit.ParticipantIdentity { return d.subscriberIdentity }
+
+func (d *DownTrack) SubscriberID() livekit.ParticipantID { return d.subscriberID }
 
 // Sets RTP header extensions for this track
 func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeaderExtensionParameter) {
@@ -349,7 +364,7 @@ func (d *DownTrack) Stop() error {
 	if d.transceiver != nil {
 		return d.transceiver.Stop()
 	}
-	return fmt.Errorf("d.transceiver not exists")
+	return errors.New("downtrack transceiver does not exist")
 }
 
 func (d *DownTrack) SetTransceiver(transceiver *webrtc.RTPTransceiver) {
@@ -613,6 +628,10 @@ func (d *DownTrack) Mute(muted bool) {
 	}
 }
 
+func (d *DownTrack) IsClosed() bool {
+	return d.isClosed.Load()
+}
+
 func (d *DownTrack) Close() {
 	d.CloseWithFlush(true)
 }
@@ -624,50 +643,52 @@ func (d *DownTrack) Close() {
 // 2. in case of session migration, participant migrate from other node, video track should
 //    be resumed with same participant, set flush=false since we don't need to flush decoder.
 func (d *DownTrack) CloseWithFlush(flush bool) {
-	if !d.bound.Load() {
+	if d.isClosed.Swap(true) {
+		// already closed
 		return
 	}
-	if d.forwarder != nil {
-		d.forwarder.Mute(true)
+
+	d.logger.Infow("close down track", "flushBlankFrame", flush)
+	if d.bound.Load() {
+		if d.forwarder != nil {
+			d.forwarder.Mute(true)
+		}
+		// write blank frames after disabling so that other frames do not interfere.
+		// Idea here is to send blank key frames to flush the decoder buffer at the remote end.
+		// Otherwise, with transceiver re-use last frame from previous stream is held in the
+		// display buffer and there could be a brief moment where the previous stream is displayed.
+		if flush {
+			doneFlushing := d.writeBlankFrameRTP(RTPBlankFramesCloseSeconds, d.blankFramesGeneration.Inc())
+
+			// wait a limited time to flush
+			timer := time.NewTimer(flushTimeout)
+			defer timer.Stop()
+
+			select {
+			case <-doneFlushing:
+			case <-timer.C:
+				d.blankFramesGeneration.Inc() // in case flush is still running
+			}
+		}
+
+		d.bound.Store(false)
+		d.logger.Debugw("closing sender", "kind", d.kind)
+		d.receiver.DeleteDownTrack(d.subscriberID)
 	}
 
-	// write blank frames after disabling so that other frames do not interfere.
-	// Idea here is to send blank key frames to flush the decoder buffer at the remote end.
-	// Otherwise, with transceiver re-use last frame from previous stream is held in the
-	// display buffer and there could be a brief moment where the previous stream is displayed.
-	d.logger.Infow("close down track", "peerID", d.peerID, "trackID", d.id, "flushBlankFrame", flush)
-	if flush {
-		doneFlushing := d.writeBlankFrameRTP(RTPBlankFramesCloseSeconds, d.blankFramesGeneration.Inc())
+	d.connectionStats.Close()
+	d.rtpStats.Stop()
+	d.logger.Debugw("rtp stats", "stats", d.rtpStats.ToString())
 
-		// wait a limited time to flush
-		timer := time.NewTimer(flushTimeout)
-		defer timer.Stop()
-
-		select {
-		case <-doneFlushing:
-		case <-timer.C:
-			d.blankFramesGeneration.Inc() // in case flush is still running
-		}
+	if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo {
+		d.onMaxLayerChanged(d, InvalidLayerSpatial)
 	}
 
-	d.closeOnce.Do(func() {
-		d.logger.Debugw("closing sender", "peerID", d.peerID, "trackID", d.id, "kind", d.kind)
-		d.receiver.DeleteDownTrack(d.peerID)
+	if d.onCloseHandler != nil {
+		d.onCloseHandler()
+	}
 
-		d.connectionStats.Close()
-		d.rtpStats.Stop()
-		d.logger.Debugw("rtp stats", "stats", d.rtpStats.ToString())
-
-		if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo {
-			d.onMaxLayerChanged(d, InvalidLayerSpatial)
-		}
-
-		if d.onCloseHandler != nil {
-			d.onCloseHandler()
-		}
-
-		d.stopKeyFrameRequester()
-	})
+	d.stopKeyFrameRequester()
 }
 
 func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) {
@@ -1349,7 +1370,8 @@ func (d *DownTrack) DebugInfo() map[string]interface{} {
 	}
 
 	return map[string]interface{}{
-		"PeerID":              d.peerID,
+		"SubscriberID":        d.subscriberID,
+		"Subscriber":          d.subscriberIdentity,
 		"TrackID":             d.id,
 		"StreamID":            d.streamID,
 		"SSRC":                d.ssrc,

pkg/sfu/downtrackspreader.go

@@ -61,23 +61,23 @@ func (d *DownTrackSpreader) Store(ts TrackSender) {
 	d.downTrackMu.Lock()
 	defer d.downTrackMu.Unlock()
 
-	d.downTracks[ts.PeerID()] = ts
+	d.downTracks[ts.SubscriberID()] = ts
 	d.shadowDownTracks()
 }
 
-func (d *DownTrackSpreader) Free(peerID livekit.ParticipantID) {
+func (d *DownTrackSpreader) Free(subscriberID livekit.ParticipantID) {
 	d.downTrackMu.Lock()
 	defer d.downTrackMu.Unlock()
 
-	delete(d.downTracks, peerID)
+	delete(d.downTracks, subscriberID)
 	d.shadowDownTracks()
 }
 
-func (d *DownTrackSpreader) HasDownTrack(peerID livekit.ParticipantID) bool {
+func (d *DownTrackSpreader) HasDownTrack(subscriberID livekit.ParticipantID) bool {
 	d.downTrackMu.RLock()
 	defer d.downTrackMu.RUnlock()
 
-	_, ok := d.downTracks[peerID]
+	_, ok := d.downTracks[subscriberID]
 	return ok
 }
 

pkg/sfu/receiver.go

@@ -49,7 +49,7 @@ type TrackReceiver interface {
 	SetMaxExpectedSpatialLayer(layer int32)
 
 	AddDownTrack(track TrackSender) error
-	DeleteDownTrack(peerID livekit.ParticipantID)
+	DeleteDownTrack(participantID livekit.ParticipantID)
 
 	DebugInfo() map[string]interface{}
 
@@ -63,7 +63,6 @@ type WebRTCReceiver struct {
 	pliThrottleConfig config.PLIThrottleConfig
 	audioConfig       config.AudioConfig
 
-	peerID         livekit.ParticipantID
 	trackID        livekit.TrackID
 	streamID       string
 	kind           webrtc.RTPCodecType
@@ -163,7 +162,6 @@ func WithLoadBalanceThreshold(downTracks int) ReceiverOpts {
 func NewWebRTCReceiver(
 	receiver *webrtc.RTPReceiver,
 	track *webrtc.TrackRemote,
-	pid livekit.ParticipantID,
 	trackInfo *livekit.TrackInfo,
 	logger logger.Logger,
 	twcc *twcc.Responder,
@@ -171,7 +169,6 @@ func NewWebRTCReceiver(
 ) *WebRTCReceiver {
 	w := &WebRTCReceiver{
 		logger:   logger,
-		peerID:   pid,
 		receiver: receiver,
 		trackID:  livekit.TrackID(track.ID()),
 		streamID: track.StreamID(),
@@ -381,7 +378,7 @@ func (w *WebRTCReceiver) AddDownTrack(track TrackSender) error {
 		return ErrReceiverClosed
 	}
 
-	if w.downTrackSpreader.HasDownTrack(track.PeerID()) {
+	if w.downTrackSpreader.HasDownTrack(track.SubscriberID()) {
 		return ErrDownTrackAlreadyExist
 	}
 
@@ -423,12 +420,12 @@ func (w *WebRTCReceiver) OnCloseHandler(fn func()) {
 }
 
 // DeleteDownTrack removes a DownTrack from a Receiver
-func (w *WebRTCReceiver) DeleteDownTrack(peerID livekit.ParticipantID) {
+func (w *WebRTCReceiver) DeleteDownTrack(subscriberID livekit.ParticipantID) {
 	if w.closed.Load() {
 		return
 	}
 
-	w.downTrackSpreader.Free(peerID)
+	w.downTrackSpreader.Free(subscriberID)
 }
 
 func (w *WebRTCReceiver) sendRTCP(packets []rtcp.Packet) {