diff --git a/pkg/sfu/redreceiver.go b/pkg/sfu/redreceiver.go index b814940f4..0d2876cf9 100644 --- a/pkg/sfu/redreceiver.go +++ b/pkg/sfu/redreceiver.go @@ -106,9 +106,10 @@ func (r *RedReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) { } func (r *RedReceiver) encodeRedForPrimary(pkt *rtp.Packet, redPayload []byte) (int, error) { - redPkts := make([]*rtp.Packet, 0, maxRedCount+1) + redLength := len(r.pktBuff) + redPkts := make([]*rtp.Packet, 0, redLength+1) lastNilPkt := -1 - for i := len(r.pktBuff) - 1; i >= 0; i-- { + for i := redLength - 1; i >= 0; i-- { if r.pktBuff[i] == nil { lastNilPkt = i break @@ -118,17 +119,26 @@ func (r *RedReceiver) encodeRedForPrimary(pkt *rtp.Packet, redPayload []byte) (i for _, prev := range r.pktBuff[lastNilPkt+1:] { if pkt.SequenceNumber == prev.SequenceNumber || - (pkt.SequenceNumber-prev.SequenceNumber) > uint16(maxRedCount) { + (pkt.SequenceNumber-prev.SequenceNumber) > uint16(redLength) || + (pkt.Timestamp-prev.Timestamp) >= (1<<14) { continue } redPkts = append(redPkts, prev) } - if r.pktBuff[1] == nil || pkt.SequenceNumber-r.pktBuff[1].SequenceNumber < 0x8000 { - /* update packet, not copy the rtp packet here since we only hold two packets for red encoding, - the upstream buffer size is much larger than two, so it is safe to use packet directly - */ - r.pktBuff[0], r.pktBuff[1] = r.pktBuff[1], pkt + // 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. + 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 + // age out older ones + for j := 0; j < i; j++ { + r.pktBuff[j] = r.pktBuff[j+1] + } + r.pktBuff[i] = pkt + break + } } return encodeRedForPrimary(redPkts, pkt, redPayload) diff --git a/pkg/sfu/redreceiver_test.go b/pkg/sfu/redreceiver_test.go index 65d273773..72d8d9a0e 100644 --- a/pkg/sfu/redreceiver_test.go +++ b/pkg/sfu/redreceiver_test.go @@ -95,8 +95,8 @@ func TestRedReceiver(t *testing.T) { }, 0) verifyRedEncodings(t, dt.lastReceivedPkt, expectPkt) } - }) + t.Run("unorder and repeat", func(t *testing.T) { w := &WebRTCReceiver{kind: webrtc.RTPCodecTypeAudio} red := w.GetRedReceiver().(*RedReceiver) @@ -126,7 +126,7 @@ func TestRedReceiver(t *testing.T) { verifyRedEncodings(t, dt.lastReceivedPkt, expectPkt) }) - t.Run("encoding excceed space", func(t *testing.T) { + t.Run("encoding exceed space", func(t *testing.T) { w := &WebRTCReceiver{isRED: true, kind: webrtc.RTPCodecTypeAudio} require.Equal(t, w.GetRedReceiver(), w) w.isRED = false @@ -146,6 +146,37 @@ func TestRedReceiver(t *testing.T) { verifyRedEncodings(t, dt.lastReceivedPkt, expectPkt) } }) + + t.Run("large timestamp gap", func(t *testing.T) { + w := &WebRTCReceiver{isRED: true, kind: webrtc.RTPCodecTypeAudio} + require.Equal(t, w.GetRedReceiver(), w) + w.isRED = false + red := w.GetRedReceiver().(*RedReceiver) + require.NotNil(t, red) + require.NoError(t, red.AddDownTrack(dt)) + + header := rtp.Header{SequenceNumber: 65534, Timestamp: (uint32(1) << 31) - 2*tsStep, PayloadType: 111} + // first few packets normal + expectPkt := make([]*rtp.Packet, 0, maxRedCount+1) + for _, pkt := range generatePkts(header, 4, tsStep) { + expectPkt = append(expectPkt, pkt) + if len(expectPkt) > maxRedCount+1 { + expectPkt = expectPkt[1:] + } + red.ForwardRTP(&buffer.ExtPacket{ + Packet: pkt, + }, 0) + verifyRedEncodings(t, dt.lastReceivedPkt, expectPkt) + } + + // and then a few packets with a large timestmap jump, should contain only primary + for _, pkt := range generatePkts(header, 4, 40*tsStep) { + red.ForwardRTP(&buffer.ExtPacket{ + Packet: pkt, + }, 0) + verifyRedEncodings(t, dt.lastReceivedPkt, []*rtp.Packet{pkt}) + } + }) } func verifyRedEncodings(t *testing.T, red *rtp.Packet, redPkts []*rtp.Packet) {