diff --git a/pkg/sfu/buffer/frameintegrity.go b/pkg/sfu/buffer/frameintegrity.go index 935263b7d..e2cbe8a43 100644 --- a/pkg/sfu/buffer/frameintegrity.go +++ b/pkg/sfu/buffer/frameintegrity.go @@ -100,7 +100,15 @@ func (ph *PacketHistory) AddPacket(extSeq uint64) { return } - for i := ph.last + 1; i < extSeq; i++ { + // A forward jump only needs at most packetCount slots cleared; anything older aliases + // into the same ring and would be overwritten anyway. Cap the loop so a single crafted + // sequence number jump (up to ~32k by the extension wrap-around heuristic) cannot force + // a large per-packet iteration count. + start := ph.last + 1 + if extSeq-start > uint64(ph.packetCount) { + start = extSeq - uint64(ph.packetCount) + } + for i := start; i < extSeq; i++ { ph.set(i, false) } @@ -208,8 +216,13 @@ func (fc *FrameIntegrityChecker) AddPacket(extSeq uint64, extFrameNum uint64, dd return } - // reset missing frames - for i := fc.last + 1; i <= extFrameNum; i++ { + // reset missing frames; cap to frameCount so a crafted frame-number jump cannot force a + // large loop (older frames alias into the same ring and get overwritten anyway). + start := fc.last + 1 + if extFrameNum-fc.last > uint64(fc.frameCount) { + start = extFrameNum - uint64(fc.frameCount) + 1 + } + for i := start; i <= extFrameNum; i++ { fc.frames[int(i-fc.base)%fc.frameCount].Reset() } fc.frames[int(extFrameNum-fc.base)%fc.frameCount].AddPacket(extSeq, ddVal) diff --git a/pkg/sfu/buffer/frameintegrity_test.go b/pkg/sfu/buffer/frameintegrity_test.go index 2815cf5e9..c492e3d91 100644 --- a/pkg/sfu/buffer/frameintegrity_test.go +++ b/pkg/sfu/buffer/frameintegrity_test.go @@ -15,6 +15,7 @@ package buffer import ( + "math/bits" "math/rand" "testing" @@ -72,8 +73,8 @@ func TestFrameIntegrityChecker(t *testing.T) { frames = append(frames, i) } require.False(t, fc.FrameIntegrity(frame)) - rand.Seed(int64(frame)) - rand.Shuffle(len(frames), func(i, j int) { frames[i], frames[j] = frames[j], frames[i] }) + rng := rand.New(rand.NewSource(int64(frame))) + rng.Shuffle(len(frames), func(i, j int) { frames[i], frames[j] = frames[j], frames[i] }) for i, f := range frames { fc.AddPacket(f, frame, &dd.DependencyDescriptor{ FirstPacketInFrame: f == firstFrame, @@ -84,3 +85,63 @@ func TestFrameIntegrityChecker(t *testing.T) { require.True(t, fc.FrameIntegrity(frame)) } } + +func countSetBits(ph *PacketHistory) int { + n := 0 + for _, w := range ph.bits { + n += bits.OnesCount64(w) + } + return n +} + +// A forward sequence-number jump much larger than the ring must clear the whole ring, +// leaving only the newly received sequence number set. +func TestPacketHistoryLargeForwardJump(t *testing.T) { + ph := NewPacketHistory(1000) // rounds up to a multiple of 64 + require.Equal(t, 1024, ph.packetCount) + + // Fill the entire ring so every slot holds a "received" bit. + base := uint64(100000) + ph.AddPacket(base) + for i := base + 1; i <= base+2000; i++ { + ph.AddPacket(i) + } + require.Equal(t, ph.packetCount, countSetBits(ph)) + last := base + 2000 + + // Forward jump well beyond both the ring and the ~32k extension wrap-around cap. The ring + // must end up fully cleared, with only newSeq marked received. + newSeq := last + 40000 + ph.AddPacket(newSeq) + + // If the cap under-cleared, stale bits from the pre-jump fill would survive here. + require.Equal(t, 1, countSetBits(ph)) + require.True(t, ph.PacketsConsecutive(newSeq, newSeq)) + require.False(t, ph.PacketsConsecutive(newSeq-5, newSeq)) + + // The window just below newSeq was cleared and can be refilled normally. + for i := newSeq - 5; i < newSeq; i++ { + ph.AddPacket(i) + } + require.True(t, ph.PacketsConsecutive(newSeq-5, newSeq)) +} + +// A forward frame-number jump much larger than frameCount must reset the whole frame ring, +// so no frame that aliases an old slot inherits stale integrity. +func TestFrameIntegrityCheckerLargeFrameJump(t *testing.T) { + fc := NewFrameIntegrityChecker(100, 1000) + + // Populate every ring slot with an integral single-packet frame. + for f := uint64(200); f <= 399; f++ { + fc.AddPacket(f, f, &dd.DependencyDescriptor{FirstPacketInFrame: true, LastPacketInFrame: true}) + } + require.True(t, fc.FrameIntegrity(399)) + + // Jump far beyond frameCount. The capped reset loop must clear the entire frame ring; if it + // under-cleared, some aliased slot would still report a stale frame's integrity. + newFrame := uint64(399 + 5000) + fc.AddPacket(50000, newFrame, &dd.DependencyDescriptor{}) // incomplete frame, no first/last + for f := newFrame - uint64(fc.frameCount) + 1; f <= newFrame; f++ { + require.False(t, fc.FrameIntegrity(f), "frame %d should not be integral after jump", f) + } +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index dc2bb15eb..b0326f4a9 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -714,6 +714,7 @@ func (d *DownTrack) handleUpstreamCodecChange(mimeType string) { receiver := d.Receiver() d.forwarder.Restart() + d.flushSequencer() d.forwarder.DetermineCodec(codec.RTPCodecCapability, receiver.HeaderExtensions(), receiver.VideoLayerMode()) d.connectionStats.UpdateCodec(d.Mime(), isFECEnabled) @@ -1034,6 +1035,15 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { return 0 } + if tp.incomingHeaderSize > len(extPkt.Packet.Payload) { + d.params.Logger.Errorw( + "incoming header size overflow", errPayloadOverflow, + "incomingHeaderSize", tp.incomingHeaderSize, + "payloadSize", len(extPkt.Packet.Payload), + ) + return 0 + } + poolEntity := PacketFactory.Get().(*[]byte) payload := *poolEntity copy(payload, tp.codecBytes) @@ -1717,6 +1727,16 @@ func (d *DownTrack) Pause() VideoAllocation { func (d *DownTrack) Resync() { d.forwarder.Resync() + d.flushSequencer() +} + +// flushSequencer discards recorded packet metadata on a stream restart so that NACK +// retransmissions cannot use metadata that describes packets no longer in the receiver's +// (resynced) bucket. +func (d *DownTrack) flushSequencer() { + if d.sequencer != nil { + d.sequencer.flush() + } } func (d *DownTrack) ReceiverRestart(rcvr TrackReceiver) { @@ -1732,6 +1752,7 @@ func (d *DownTrack) ReceiverRestart(rcvr TrackReceiver) { receiver := d.Receiver() d.params.Logger.Infow("upstream receiver restart", "mime", receiver.Mime().String()) d.forwarder.Restart() + d.flushSequencer() d.forwarder.DetermineCodec(codec, receiver.HeaderExtensions(), receiver.VideoLayerMode()) } @@ -2157,6 +2178,18 @@ func (d *DownTrack) retransmitPacket(epm *extPacketMeta, sourcePkt []byte, isPro d.params.Logger.Errorw("could not unmarshal rtp packet to send via RTX", err) return 0, err } + // Defensive panic-safety net: the codec header size was recorded when the packet was first + // forwarded and is re-read here against a bucket packet. A stream restart flushes the + // sequencer (see flushSequencer), so metadata and payload should always agree; guard + // against a slice overflow regardless. + if int(epm.numCodecBytesIn) > len(pkt.Payload) { + d.params.Logger.Warnw( + "recorded codec header size overflows payload", errPayloadOverflow, + "numCodecBytesIn", epm.numCodecBytesIn, + "payloadSize", len(pkt.Payload), + ) + return 0, errPayloadOverflow + } hdr := RTPHeaderFactory.Get().(*rtp.Header) *hdr = rtp.Header{ Version: pkt.Header.Version, diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index 1e08ef9ce..0507120c1 100644 --- a/pkg/sfu/sequencer.go +++ b/pkg/sfu/sequencer.go @@ -148,6 +148,26 @@ func (s *sequencer) setRTT(rtt uint32) { } } +// flush discards all recorded packet metadata. It must be called on a stream restart: the +// metadata maps outgoing sequence numbers to source packets in the receiver's bucket, and a +// restart resyncs that bucket, so retransmitting against stale metadata would send the wrong +// packet (or read past the re-read payload). After flush, NACKs for pre-restart packets are +// ignored until the sequencer is re-initialized by the next push. +func (s *sequencer) flush() { + s.Lock() + defer s.Unlock() + + s.initialized = false + s.extStartSN = 0 + s.extHighestSN = 0 + s.extHighestTS = 0 + s.snOffset = 0 + clear(s.meta) + if s.snRangeMap != nil { + s.snRangeMap = utils.NewRangeMap[uint64, uint64]((s.size + 1) / 2) + } +} + func (s *sequencer) push( packetTime int64, extIncomingSN, extModifiedSN uint64, diff --git a/pkg/sfu/sequencer_test.go b/pkg/sfu/sequencer_test.go index 862ac531b..be91af8be 100644 --- a/pkg/sfu/sequencer_test.go +++ b/pkg/sfu/sequencer_test.go @@ -78,6 +78,39 @@ func Test_sequencer(t *testing.T) { require.Equal(t, 1, len(m)) } +func Test_sequencer_flush(t *testing.T) { + seq := newSequencer(500, false, logger.GetLogger()) + off := uint16(15) + + for i := uint64(1); i < 100; i++ { + seq.push(time.Now().UnixNano(), i, i+uint64(off), 123, true, 2, nil, 0, nil, nil) + } + preFlush := []uint16{57 + off, 58 + off} + + // flush discards all recorded metadata on a stream restart + seq.flush() + + // even after enough time elapses, a NACK for a pre-flush packet retransmits nothing + time.Sleep((ignoreRetransmission + 10) * time.Millisecond) + require.Equal(t, 0, len(seq.getExtPacketMetas(preFlush))) + + // the sequencer re-initializes on the next push and works normally for new packets + for i := uint64(200); i < 210; i++ { + seq.push(time.Now().UnixNano(), i, i+uint64(off), 456, true, 3, nil, 0, nil, nil) + } + postFlush := []uint16{205 + off} + require.Equal(t, 0, len(seq.getExtPacketMetas(postFlush))) // not enough time elapsed yet + time.Sleep((ignoreRetransmission + 10) * time.Millisecond) + res := seq.getExtPacketMetas(postFlush) + require.Equal(t, 1, len(res)) + require.Equal(t, uint16(205+off), res[0].targetSeqNo) + require.Equal(t, uint64(205), res[0].sourceSeqNo) + require.Equal(t, int8(3), res[0].layer) + + // pre-flush packets remain non-retransmittable + require.Equal(t, 0, len(seq.getExtPacketMetas(preFlush))) +} + func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) { type args struct { seqNo []uint16