fix: copy RED history packets instead of keeping pointers (#4941)

RedReceiver kept pointers to the last forwarded packets for its
redundant blocks, but their payload is in the read buffer that
forwardRTP reuses for every packet, so the blocks carried the current
packet's bytes. Copy the history packets into slots RedReceiver owns.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

pkg/sfu/redreceiver.go

@@ -17,6 +17,7 @@ package sfu
 import (
 	"encoding/binary"
 	"fmt"
+	"slices"
 
 	"go.uber.org/atomic"
 
@@ -50,7 +51,11 @@ type RedReceiver struct {
 	logger            logger.Logger
 	closed            atomic.Bool
 	pktBuff           [maxRedCount]*rtp.Packet
-	redPayloadBuf     [mtuSize]byte
+	// copies that pktBuff points at, as a forwarded packet's payload is in a read buffer reused for the next packet,
+	// one more slot than pktBuff so that a new copy never overwrites a redundant block of the packet being encoded
+	histPkts      [maxRedCount + 1]rtp.Packet
+	histPayloads  [maxRedCount + 1][maxRedPayload]byte
+	redPayloadBuf [mtuSize]byte
 	// forwarded packet, reused like redPayloadBuf since ForwardRTP runs on one goroutine
 	// and down tracks do not keep the packet past WriteRTP
 	redExtPkt buffer.ExtPacket
@@ -189,17 +194,22 @@ func (r *RedReceiver) encodeRedForPrimary(pkt *rtp.Packet, redPayload []byte) (i
 		redPkts = append(redPkts, prev)
 	}
 
-	// insert primary packet in history buffer
-	// NOTE: packet is copied from retransmission buffer and used in forwarding path. So, not making another
-	// copy here and just maintaining pointer to the packet as the forwarding path should not alter the packet.
+	// insert a copy of the primary packet in history buffer
 	for i := redLength - 1; i >= 0; i-- {
 		if r.pktBuff[i] == nil || // history is empty
 			pkt.SequenceNumber-r.pktBuff[i].SequenceNumber < (1<<15) { // received packet has more recent sequence number
+			// pick the slot before aging out, as the aged out packet can be a redundant block of this packet
+			slot := r.freeHistSlot()
 			// age out older ones
 			for j := 0; j < i; j++ {
 				r.pktBuff[j] = r.pktBuff[j+1]
 			}
-			r.pktBuff[i] = pkt
+			n := copy(r.histPayloads[slot][:], pkt.Payload)
+			r.histPkts[slot] = rtp.Packet{
+				Header:  rtp.Header{SequenceNumber: pkt.SequenceNumber, Timestamp: pkt.Timestamp},
+				Payload: r.histPayloads[slot][:n],
+			}
+			r.pktBuff[i] = &r.histPkts[slot]
 			break
 		}
 	}
@@ -207,6 +217,16 @@ func (r *RedReceiver) encodeRedForPrimary(pkt *rtp.Packet, redPayload []byte) (i
 	return encodeRedForPrimary(redPkts, pkt, redPayload)
 }
 
+// freeHistSlot returns a history slot that pktBuff does not point at, there is always one as there is a spare slot
+func (r *RedReceiver) freeHistSlot() int {
+	for slot := range r.histPkts {
+		if !slices.Contains(r.pktBuff[:], &r.histPkts[slot]) {
+			return slot
+		}
+	}
+	return 0
+}
+
 func encodeRedForPrimary(redPkts []*rtp.Packet, primary *rtp.Packet, redPayload []byte) (int, error) {
 	payloadSize := len(primary.Payload) + 1
 	for _, p := range redPkts {

pkg/sfu/redreceiver_test.go

@@ -15,6 +15,7 @@
 package sfu
 
 import (
+	"bytes"
 	"testing"
 
 	"github.com/pion/rtp"
@@ -22,6 +23,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+	"github.com/livekit/mediatransportutil/pkg/bucket"
 	"github.com/livekit/protocol/logger"
 )
 
@@ -234,6 +236,55 @@ func TestRedReceiver(t *testing.T) {
 	})
 }
 
+// reads every packet into one reused buffer, the way ReceiverBase.forwardRTP does
+func TestRedReceiverReusedReadBuffer(t *testing.T) {
+	opus := webrtc.RTPCodecParameters{
+		RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2},
+		PayloadType:        111,
+	}
+	buff := buffer.NewBuffer(0x5678, 500, 200)
+	require.NoError(t, buff.Bind(webrtc.RTPParameters{Codecs: []webrtc.RTPCodecParameters{opus}}, opus.RTPCodecCapability, 0))
+
+	w := &WebRTCReceiver{
+		ReceiverBase: &ReceiverBase{
+			params: ReceiverBaseParams{
+				Kind:   webrtc.RTPCodecTypeAudio,
+				Logger: logger.GetLogger(),
+			},
+		},
+	}
+	red := w.GetRedReceiver().(*RedReceiver)
+	dt := &dummyDowntrack{TrackSender: &DownTrack{}}
+	require.NoError(t, red.AddDownTrack(dt))
+
+	var payloads [][]byte
+	pktBuf := make([]byte, bucket.RTPMaxPktSize)
+	for i := range 6 {
+		payload := bytes.Repeat([]byte{byte(0x10 * (i + 1))}, 60)
+		payloads = append(payloads, payload)
+		pkt := rtp.Packet{
+			Header:  rtp.Header{Version: 2, PayloadType: 111, SequenceNumber: uint16(100 + i), Timestamp: uint32(960 * i), SSRC: 0x5678},
+			Payload: payload,
+		}
+		raw, err := pkt.Marshal()
+		require.NoError(t, err)
+		_, err = buff.Write(raw)
+		require.NoError(t, err)
+		extPkt, err := buff.ReadExtended(pktBuf)
+		require.NoError(t, err)
+		red.ForwardRTP(extPkt, 0)
+		buffer.ReleaseExtPacket(extPkt)
+	}
+
+	// the last RED packet carries the two packets before it as redundant blocks
+	blocks, err := extractPktsFromRed(dt.lastReceivedPkt, 0xFF)
+	require.NoError(t, err)
+	require.Len(t, blocks, 3)
+	for i, block := range blocks {
+		require.Equal(t, payloads[3+i], block.Payload)
+	}
+}
+
 func verifyRedEncodings(t *testing.T, red *rtp.Packet, redPkts []*rtp.Packet) {
 	solidPkts := make([]*rtp.Packet, 0, len(redPkts))
 	for _, pkt := range redPkts {