diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 49eea11d4..cca32e2e8 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -123,12 +123,13 @@ type StreamInfoProbe struct { func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer { b := &Buffer{} b.BufferBase = NewBufferBase(BufferBaseParams{ - SSRC: ssrc, - MaxVideoPkts: maxVideoPkts, - MaxAudioPkts: maxAudioPkts, - LoggerComponents: []string{sutils.ComponentPub, sutils.ComponentSFU}, - SendPLI: b.sendPLI, - IsReportingEnabled: true, + SSRC: ssrc, + MaxVideoPkts: maxVideoPkts, + MaxAudioPkts: maxAudioPkts, + LoggerComponents: []string{sutils.ComponentPub, sutils.ComponentSFU}, + SendPLI: b.sendPLI, + OnStreamRestartLocked: b.recreateFECDecoderLocked, + IsReportingEnabled: true, }) return b } @@ -489,6 +490,19 @@ func (b *Buffer) maybeCreateFECDecoderLocked() { b.logger.Debugw("flexfec decoder created", "fecSSRC", b.fecSSRC, "mediaSSRC", b.BufferBase.SSRC()) } +// recreateFECDecoderLocked drops FlexFEC state from the previous sequence +// space. Stream restart resyncs the packet bucket and sequence map, so repair +// packets retained by the old decoder would recover against the wrong media +// or be treated as duplicates of the new repair stream. +func (b *Buffer) recreateFECDecoderLocked() { + if b.fecDecoder == nil { + return + } + + b.fecDecoder = nil + b.maybeCreateFECDecoderLocked() +} + func (b *Buffer) getFECMediaPacketLocked(sequenceNumber uint16, dst []byte) (int, error) { if b.bucket == nil || b.rtpStats == nil { return 0, errFECMediaPacketNotFound diff --git a/pkg/sfu/buffer/buffer_base.go b/pkg/sfu/buffer/buffer_base.go index 8e7b33438..ca831e212 100644 --- a/pkg/sfu/buffer/buffer_base.go +++ b/pkg/sfu/buffer/buffer_base.go @@ -152,14 +152,15 @@ const ( ) type BufferBaseParams struct { - SSRC uint32 - MaxVideoPkts int - MaxAudioPkts int - LoggerComponents []string - SendPLI func() - IsReportingEnabled bool - IsOOBSequenceNumber bool - IsDDRestartEnabled bool + SSRC uint32 + MaxVideoPkts int + MaxAudioPkts int + LoggerComponents []string + SendPLI func() + OnStreamRestartLocked func() + IsReportingEnabled bool + IsOOBSequenceNumber bool + IsDDRestartEnabled bool } type BufferBase struct { @@ -599,6 +600,10 @@ func (b *BufferBase) restartStreamLocked(reason string, isDetected bool) { b.StartKeyFrameSeeder() + if f := b.params.OnStreamRestartLocked; f != nil { + f() + } + if isDetected { b.isRestartPending = true diff --git a/pkg/sfu/buffer/buffer_fec_test.go b/pkg/sfu/buffer/buffer_fec_test.go index 364f73a11..65f92adbd 100644 --- a/pkg/sfu/buffer/buffer_fec_test.go +++ b/pkg/sfu/buffer/buffer_fec_test.go @@ -655,3 +655,62 @@ func TestBufferFECNACKNotSentForRecoveredPacket(t *testing.T) { assert.False(t, nackedSNs[media[2].SequenceNumber], "arrived packet %d must not be NACKed", media[2].SequenceNumber) assert.False(t, nackedSNs[media[3].SequenceNumber], "FEC-recovered packet %d must not be NACKed", media[3].SequenceNumber) } + +func TestBufferFECDecoderRecreatedOnStreamRestart(t *testing.T) { + factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory() + + primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer) + fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) + factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) + bindFECTestBuffer(t, primary) + + media := rtptest.GenerateVP8Packets(100, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) + fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1) + require.Len(t, fecPackets, 1) + + const droppedIdx = 2 + for i := range media { + if i != droppedIdx { + writePacket(t, primary, &media[i]) + } + } + writePacket(t, fecBuff, &fecPackets[0]) + require.EqualValues(t, 1, primary.FECDecoderStats().PacketsRecovered) + + decoderBefore := primary.fecDecoder + require.NotNil(t, decoderBefore) + + // Receiver restart and detected discontinuity both land here. + primary.RestartStream("test") + + decoderAfter := primary.fecDecoder + require.NotNil(t, decoderAfter) + require.NotSame(t, decoderBefore, decoderAfter) + stats := primary.FECDecoderStats() + assert.Zero(t, stats.FECPacketsReceived) + assert.Zero(t, stats.PacketsRecovered) + assert.Zero(t, stats.FECPacketsDiscarded) + + // A new encoder reuses the repair sequence number of the pre-restart FEC + // packet. The old decoder would discard it as a duplicate. + media2 := rtptest.GenerateVP8Packets(500, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) + fecPackets2 := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media2, 1) + require.Len(t, fecPackets2, 1) + require.Equal(t, fecPackets[0].SequenceNumber, fecPackets2[0].SequenceNumber) + + const droppedIdx2 = 1 + for i := range media2 { + if i != droppedIdx2 { + writePacket(t, primary, &media2[i]) + } + } + writePacket(t, fecBuff, &fecPackets2[0]) + + stats = primary.FECDecoderStats() + assert.EqualValues(t, 1, stats.FECPacketsReceived) + assert.EqualValues(t, 1, stats.PacketsRecovered) + assert.EqualValues(t, 0, stats.FECPacketsDiscarded) + + extSNBySN := readExtSequenceNumbers(t, primary, len(media2)-1) + requireRecoveredInBucket(t, primary, &media2[droppedIdx2], extSNBySN, media2[0].SequenceNumber) +}