Flush sequencer on stream restart; bound frame-integrity loops (#4760)

Flush the downtrack sequencer on stream restart (Resync, ReceiverRestart,
codec change) so NACK retransmissions can't use metadata that no longer
matches the resynced bucket. Add a defensive bounds guard on the RTX and
forward payload slicing.

Cap the PacketHistory and FrameIntegrityChecker catch-up loops to the ring
size so a large sequence/frame-number jump can't drive a big per-packet
iteration count.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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)

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)
+	}
+}

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,

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,

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