diff --git a/pkg/sfu/flexfec/decoder.go b/pkg/sfu/flexfec/decoder.go index 80bca306c..50ca5b344 100644 --- a/pkg/sfu/flexfec/decoder.go +++ b/pkg/sfu/flexfec/decoder.go @@ -191,9 +191,16 @@ func (d *Decoder) insertMediaPacket(receivedPkt *rtp.Packet) { } d.recoveredPackets = append(d.recoveredPackets, receivedPkt) - sort.Slice(d.recoveredPackets, func(i, j int) bool { - return isNewerSeq(d.recoveredPackets[i].SequenceNumber, d.recoveredPackets[j].SequenceNumber) - }) + if len(d.recoveredPackets) > 1 && !isNewerSeq( + d.recoveredPackets[len(d.recoveredPackets)-2].SequenceNumber, + receivedPkt.SequenceNumber, + ) { + insertAt := sort.Search(len(d.recoveredPackets)-1, func(i int) bool { + return isNewerSeq(receivedPkt.SequenceNumber, d.recoveredPackets[i].SequenceNumber) + }) + copy(d.recoveredPackets[insertAt+1:], d.recoveredPackets[insertAt:len(d.recoveredPackets)-1]) + d.recoveredPackets[insertAt] = receivedPkt + } d.updateCoveringFecPackets(receivedPkt) } @@ -296,15 +303,22 @@ func (d *Decoder) insertFECPacket(fecPkt *rtp.Packet) { return } - d.receivedFECPackets = append(d.receivedFECPackets, fecPacketState{ + state := fecPacketState{ packet: ownedFECPkt, flexFec: ownedFEC, protectedPackets: protectedPackets, - }) - - sort.Slice(d.receivedFECPackets, func(i, j int) bool { - return isNewerSeq(d.receivedFECPackets[i].packet.SequenceNumber, d.receivedFECPackets[j].packet.SequenceNumber) - }) + } + d.receivedFECPackets = append(d.receivedFECPackets, state) + if len(d.receivedFECPackets) > 1 && !isNewerSeq( + d.receivedFECPackets[len(d.receivedFECPackets)-2].packet.SequenceNumber, + state.packet.SequenceNumber, + ) { + insertAt := sort.Search(len(d.receivedFECPackets)-1, func(i int) bool { + return isNewerSeq(state.packet.SequenceNumber, d.receivedFECPackets[i].packet.SequenceNumber) + }) + copy(d.receivedFECPackets[insertAt+1:], d.receivedFECPackets[insertAt:len(d.receivedFECPackets)-1]) + d.receivedFECPackets[insertAt] = state + } if len(d.receivedFECPackets) > maxFECPackets { d.removeFECPacketAt(0) @@ -337,12 +351,7 @@ func (d *Decoder) attemptRecovery() []*rtp.Packet { d.removeFECPacketAt(i) recoveredPackets = append(recoveredPackets, recovered) - d.recoveredPackets = append(d.recoveredPackets, recovered) - sort.Slice(d.recoveredPackets, func(i, j int) bool { - return isNewerSeq(d.recoveredPackets[i].SequenceNumber, d.recoveredPackets[j].SequenceNumber) - }) - - d.updateCoveringFecPackets(recovered) + d.insertMediaPacket(recovered) d.discardOldRecoveredPackets() packetsRecovered++ } diff --git a/pkg/sfu/flexfec/decoder_test.go b/pkg/sfu/flexfec/decoder_test.go index 622b5870d..9c55aca4f 100644 --- a/pkg/sfu/flexfec/decoder_test.go +++ b/pkg/sfu/flexfec/decoder_test.go @@ -187,6 +187,39 @@ func TestDecoderSequenceNumberWrap(t *testing.T) { requirePacketEqual(t, &media[3], recovered[0]) } +func TestDecoderMediaWindowOrder(t *testing.T) { + media := makeMediaPackets(t, 65534, 4) + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + + for _, i := range []int{1, 3, 0, 2} { + require.Empty(t, decoder.DecodeFec(&media[i])) + } + + require.Len(t, decoder.recoveredPackets, 4) + for i := range media { + assert.Equal(t, media[i].SequenceNumber, decoder.recoveredPackets[i].SequenceNumber) + } +} + +func TestDecoderFECWindowOrder(t *testing.T) { + media := makeMediaPackets(t, 50, 5) + fec := encodeFEC(t, media, 1)[0] + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + + for _, i := range []int{0, 3, 4} { + require.Empty(t, decoder.DecodeFec(&media[i])) + } + for _, seq := range []uint16{102, 100, 101} { + fec.SequenceNumber = seq + require.Empty(t, decoder.DecodeFec(&fec)) + } + + require.Len(t, decoder.receivedFECPackets, 3) + for i, seq := range []uint16{100, 101, 102} { + assert.Equal(t, seq, decoder.receivedFECPackets[i].packet.SequenceNumber) + } +} + func TestDecoderDiscardsForeignProtectedSSRC(t *testing.T) { media := makeMediaPackets(t, 300, 5) fec := encodeFEC(t, media, 1)