Ignore duplicate RID. (#2471)

Firefox on Windows 10 seems to be producing simulcast tracks with
duplicate RID. That causes a leak as only one buffer is processed.

Ignore duplicate rid.

NOTE: This is not perfect as the actual layer -> rid is indeterminable
at addition time. It would require looking at packets to determine the
video dimensions and match to rid/layer to figure out which one is
correct and which one is duplicate.

To simplify though, taking the first one and dropping later ones.
This could mean the correct resolution is not streamed, but that should
be okay. The leak is far more destructive.

pkg/rtc/mediatrack.go

@@ -309,7 +309,17 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 	}
 	t.lock.Unlock()
 
-	wr.(*sfu.WebRTCReceiver).AddUpTrack(track, buff)
+	if err := wr.(*sfu.WebRTCReceiver).AddUpTrack(track, buff); err != nil {
+		t.params.Logger.Warnw(
+			"adding up track failed", err,
+			"rid", track.RID(),
+			"layer", layer,
+			"ssrc", track.SSRC(),
+			"newCodec", newCodec,
+		)
+		buff.Close()
+		return false
+	}
 
 	// LK-TODO: can remove this completely when VideoLayers protocol becomes the default as it has info from client or if we decide to use TrackInfo.Simulcast
 	if t.numUpTracks.Inc() > 1 || track.RID() != "" {

pkg/rtc/participant.go

@@ -1137,6 +1137,8 @@ func (p *ParticipantImpl) UpdateMediaRTT(rtt uint32) {
 	}
 }
 
+// ----------------------------------------------------------
+
 type AnyTransportHandler struct {
 	transport.UnimplementedHandler
 	p *ParticipantImpl
@@ -1154,6 +1156,8 @@ func (h AnyTransportHandler) OnICECandidate(c *webrtc.ICECandidate, target livek
 	return h.p.onICECandidate(c, target)
 }
 
+// ----------------------------------------------------------
+
 type PublisherTransportHandler struct {
 	AnyTransportHandler
 }
@@ -1174,6 +1178,8 @@ func (h PublisherTransportHandler) OnDataPacket(kind livekit.DataPacket_Kind, da
 	h.p.onDataMessage(kind, data)
 }
 
+// ----------------------------------------------------------
+
 type SubscriberTransportHandler struct {
 	AnyTransportHandler
 }
@@ -1190,6 +1196,8 @@ func (h SubscriberTransportHandler) OnInitialConnected() {
 	h.p.onSubscriberInitialConnected()
 }
 
+// ----------------------------------------------------------
+
 type PrimaryTransportHandler struct {
 	transport.Handler
 	p *ParticipantImpl
@@ -1204,6 +1212,8 @@ func (h PrimaryTransportHandler) OnFullyEstablished() {
 	h.p.onPrimaryTransportFullyEstablished()
 }
 
+// ----------------------------------------------------------
+
 func (p *ParticipantImpl) setupTransportManager() error {
 	p.twcc = twcc.NewTransportWideCCResponder()
 	p.twcc.OnFeedback(func(pkts []rtcp.Packet) {

pkg/sfu/buffer/buffer.go

@@ -147,7 +147,7 @@ func (b *Buffer) SetLogger(logger logger.Logger) {
 	b.Lock()
 	defer b.Unlock()
 
-	b.logger = logger.WithComponent(sutils.ComponentSFU)
+	b.logger = logger.WithComponent(sutils.ComponentSFU).WithValues("ssrc", b.mediaSSRC)
 	if b.rtpStats != nil {
 		b.rtpStats.SetLogger(b.logger)
 	}
@@ -584,7 +584,7 @@ func (b *Buffer) patchExtPacket(ep *ExtPacket, buf []byte) *ExtPacket {
 	n, err := b.getPacket(buf, ep.Packet.SequenceNumber)
 	if err != nil {
 		packetNotFoundCount := b.packetNotFoundCount.Inc()
-		if packetNotFoundCount%20 == 0 {
+		if (packetNotFoundCount-1)%20 == 0 {
 			b.logger.Warnw("could not get packet from bucket", err, "sn", ep.Packet.SequenceNumber, "headSN", b.bucket.HeadSequenceNumber(), "count", packetNotFoundCount)
 		}
 		return nil

pkg/sfu/receiver.go

@@ -42,6 +42,7 @@ var (
 	ErrReceiverClosed        = errors.New("receiver closed")
 	ErrDownTrackAlreadyExist = errors.New("DownTrack already exist")
 	ErrBufferNotFound        = errors.New("buffer not found")
+	ErrDuplicateLayer        = errors.New("duplicate layer")
 )
 
 type AudioLevelHandle func(level uint8, duration uint32)
@@ -114,11 +115,9 @@ type WebRTCReceiver struct {
 
 	bufferMu sync.RWMutex
 	buffers  [buffer.DefaultMaxLayerSpatial + 1]*buffer.Buffer
+	upTracks [buffer.DefaultMaxLayerSpatial + 1]*webrtc.TrackRemote
 	rtt      uint32
 
-	upTrackMu sync.RWMutex
-	upTracks  [buffer.DefaultMaxLayerSpatial + 1]*webrtc.TrackRemote
-
 	lbThreshold int
 
 	streamTrackerManager *StreamTrackerManager
@@ -265,9 +264,16 @@ func (w *WebRTCReceiver) OnStatsUpdate(fn func(w *WebRTCReceiver, stat *livekit.
 }
 
 func (w *WebRTCReceiver) OnMaxLayerChange(fn func(maxLayer int32)) {
-	w.upTrackMu.Lock()
+	w.bufferMu.Lock()
 	w.onMaxLayerChange = fn
-	w.upTrackMu.Unlock()
+	w.bufferMu.Unlock()
+}
+
+func (w *WebRTCReceiver) getOnMaxLayerChange() func(maxLayer int32) {
+	w.bufferMu.RLock()
+	defer w.bufferMu.RUnlock()
+
+	return w.onMaxLayerChange
 }
 
 func (w *WebRTCReceiver) GetConnectionScoreAndQuality() (float32, livekit.ConnectionQuality) {
@@ -306,10 +312,7 @@ func (w *WebRTCReceiver) TrackID() livekit.TrackID {
 	return w.trackID
 }
 
-func (w *WebRTCReceiver) SSRC(layer int) uint32 {
-	w.upTrackMu.RLock()
-	defer w.upTrackMu.RUnlock()
-
+func (w *WebRTCReceiver) ssrc(layer int) uint32 {
 	if track := w.upTracks[layer]; track != nil {
 		return uint32(track.SSRC())
 	}
@@ -328,9 +331,9 @@ func (w *WebRTCReceiver) Kind() webrtc.RTPCodecType {
 	return w.kind
 }
 
-func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buffer) {
+func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buffer) error {
 	if w.closed.Load() {
-		return
+		return ErrReceiverClosed
 	}
 
 	layer := int32(0)
@@ -369,14 +372,16 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff
 		buff.SetPLIThrottle(duration.Nanoseconds())
 	}
 
-	w.upTrackMu.Lock()
-	w.upTracks[layer] = track
-	w.upTrackMu.Unlock()
-
 	w.bufferMu.Lock()
+	if w.upTracks[layer] != nil {
+		w.bufferMu.Unlock()
+		return ErrDuplicateLayer
+	}
+	w.upTracks[layer] = track
 	w.buffers[layer] = buff
 	rtt := w.rtt
 	w.bufferMu.Unlock()
+
 	buff.SetRTT(rtt)
 	buff.SetPaused(w.streamTrackerManager.IsPaused())
 
@@ -385,6 +390,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff
 	}
 
 	go w.forwardRTP(layer)
+	return nil
 }
 
 // SetUpTrackPaused indicates upstream will not be sending any data.
@@ -471,11 +477,7 @@ func (w *WebRTCReceiver) OnMaxTemporalLayerSeenChanged(maxTemporalLayerSeen int3
 
 // StreamTrackerManagerListener.OnMaxAvailableLayerChanged
 func (w *WebRTCReceiver) OnMaxAvailableLayerChanged(maxAvailableLayer int32) {
-	w.upTrackMu.RLock()
-	onMaxLayerChange := w.onMaxLayerChange
-	w.upTrackMu.RUnlock()
-
-	if onMaxLayerChange != nil {
+	if onMaxLayerChange := w.getOnMaxLayerChange(); onMaxLayerChange != nil {
 		onMaxLayerChange(maxAvailableLayer)
 	}
 }
@@ -619,7 +621,7 @@ func (w *WebRTCReceiver) GetDeltaStats() map[uint32]*buffer.StreamStatsWithLayer
 		patched[int32(layer)] = sswl.Layers[0]
 		sswl.Layers = patched
 
-		deltaStats[w.SSRC(layer)] = sswl
+		deltaStats[w.ssrc(layer)] = sswl
 	}
 
 	return deltaStats
@@ -711,7 +713,7 @@ func (w *WebRTCReceiver) DebugInfo() map[string]interface{} {
 		"Simulcast": isSimulcast,
 	}
 
-	w.upTrackMu.RLock()
+	w.bufferMu.RLock()
 	upTrackInfo := make([]map[string]interface{}, 0, len(w.upTracks))
 	for layer, ut := range w.upTracks {
 		if ut != nil {
@@ -723,7 +725,7 @@ func (w *WebRTCReceiver) DebugInfo() map[string]interface{} {
 			})
 		}
 	}
-	w.upTrackMu.RUnlock()
+	w.bufferMu.RUnlock()
 	info["UpTracks"] = upTrackInfo
 
 	return info