Do not include packet in RED if timestamp is too far back. (#1478)

* Do not include packet in RED if timestamp is too far back.

* add test case for large timestmap jump

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)

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) {