Process NACK retransmissions in a single worker per DownTrack (#4758)

Replace the per-NACK-packet goroutine spawn in DownTrack.handleRTCP with a
single long-lived worker that coalesces pending NACKs into one retransmit
pass. Pending sequence numbers are capped so a high NACK arrival rate cannot
grow goroutine count or memory unboundedly.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

pkg/sfu/downtrack.go

@@ -65,6 +65,9 @@ const (
 	keyFrameIntervalMax = 1000
 	flushTimeout        = 1 * time.Second
 
+	// upper bound on NACKed sequence numbers buffered between retransmit worker runs
+	maxPendingNACKs = 5000
+
 	waitBeforeSendPaddingOnMute = 100 * time.Millisecond
 	maxPaddingOnMuteDuration    = 5 * time.Second
 	paddingOnMuteInterval       = 100 * time.Millisecond
@@ -336,6 +339,12 @@ type DownTrack struct {
 	keyFrameRequesterCh       chan struct{}
 	keyFrameRequesterChClosed bool
 
+	retransmitChMu     sync.RWMutex
+	retransmitCh       chan struct{}
+	retransmitChClosed bool
+	pendingNACKsMu     sync.Mutex
+	pendingNACKs       []uint16
+
 	createdAt     int64
 	lastUnmutedAt atomic.Time
 }
@@ -363,6 +372,7 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) {
 		pacer:               params.Pacer,
 		maxLayerNotifierCh:  make(chan string, 1),
 		keyFrameRequesterCh: make(chan struct{}, 1),
+		retransmitCh:        make(chan struct{}, 1),
 		createdAt:           time.Now().UnixNano(),
 		receiver:            params.Receiver,
 	}
@@ -422,6 +432,8 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) {
 		go d.keyFrameRequester()
 	}
 
+	go d.retransmitWorker()
+
 	d.params.Receiver.AddOnReady(d.handleReceiverReady)
 	d.rtxSequenceNumber.Store(uint64(rand.Intn(1<<14)) + uint64(1<<15)) // a random number in third quartile of sequence number space
 	d.params.Logger.Debugw("downtrack created", "upstreamCodecs", d.upstreamCodecs)
@@ -1461,6 +1473,11 @@ func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) {
 	close(d.keyFrameRequesterCh)
 	d.keyFrameRequesterChMu.Unlock()
 
+	d.retransmitChMu.Lock()
+	d.retransmitChClosed = true
+	close(d.retransmitCh)
+	d.retransmitChMu.Unlock()
+
 	d.params.Listener.OnDownTrackClose(!flush)
 }
 
@@ -2032,7 +2049,7 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 					numNACKs += uint32(len(packetList))
 					nacks = append(nacks, packetList...)
 				}
-				go d.retransmitPackets(nacks)
+				d.postRetransmitPackets(nacks)
 			}
 
 		case *rtcp.TransportLayerCC:
@@ -2253,6 +2270,49 @@ func (d *DownTrack) retransmitPacket(epm *extPacketMeta, sourcePkt []byte, isPro
 	return headerSize + len(payload), nil
 }
 
+// postRetransmitPackets buffers the NACKed sequence numbers, capped at
+// maxPendingNACKs, and signals the retransmit worker without blocking.
+func (d *DownTrack) postRetransmitPackets(nacks []uint16) {
+	if d.sequencer == nil || len(nacks) == 0 {
+		return
+	}
+
+	d.pendingNACKsMu.Lock()
+	d.pendingNACKs = append(d.pendingNACKs, nacks...)
+	if len(d.pendingNACKs) > maxPendingNACKs {
+		// drop the oldest NACKs under a NACK flood
+		d.pendingNACKs = d.pendingNACKs[len(d.pendingNACKs)-maxPendingNACKs:]
+	}
+	d.pendingNACKsMu.Unlock()
+
+	d.retransmitChMu.RLock()
+	if !d.retransmitChClosed {
+		select {
+		case d.retransmitCh <- struct{}{}:
+		default:
+		}
+	}
+	d.retransmitChMu.RUnlock()
+}
+
+func (d *DownTrack) retransmitWorker() {
+	for {
+		_, more := <-d.retransmitCh
+		if !more {
+			return
+		}
+
+		d.pendingNACKsMu.Lock()
+		nacks := d.pendingNACKs
+		d.pendingNACKs = nil
+		d.pendingNACKsMu.Unlock()
+
+		if len(nacks) > 0 {
+			d.retransmitPackets(nacks)
+		}
+	}
+}
+
 func (d *DownTrack) retransmitPackets(nacks []uint16) {
 	if d.sequencer == nil {
 		return