H265 supoort and codec regression (#3358)

* H265 supoort and codec regression

Support H265 codec.
Add optional codec regression for subscribers don't
support advanced codecs like H265, AV1, VP9.

* restart forwarder on upstream codec change

* tests

* Reneogitate new codec if client doesn't support change

* Add option to disable codec regression

---------

Co-authored-by: boks1971 <raja.gobi@tutanota.com>

pkg/sfu/buffer/buffer.go

@@ -17,6 +17,7 @@ package buffer
 import (
 	"encoding/binary"
 	"errors"
+	"fmt"
 	"io"
 	"strings"
 	"sync"
@@ -79,7 +80,6 @@ type Buffer struct {
 	maxVideoPkts    int
 	maxAudioPkts    int
 	codecType       webrtc.RTPCodecType
-	payloadType     uint8
 	extPackets      deque.Deque[*ExtPacket]
 	pPackets        []pendingPacket
 	closeOnce       sync.Once
@@ -90,7 +90,11 @@ type Buffer struct {
 	audioLevelExtID uint8
 	bound           bool
 	closed          atomic.Bool
-	mime            string
+
+	rtpParameters  webrtc.RTPParameters
+	payloadType    uint8
+	rtxPayloadType uint8
+	mime           string
 
 	snRangeMap *utils.RangeMap[uint64, uint64]
 
@@ -119,6 +123,7 @@ type Buffer struct {
 	onRtcpSenderReport func()
 	onFpsChanged       func()
 	onFinalRtpStats    func(*livekit.RTPStats)
+	onCodecChange      func(webrtc.RTPCodecParameters)
 
 	// logger
 	logger logger.Logger
@@ -214,6 +219,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 	b.clockRate = codec.ClockRate
 	b.lastReport = mono.UnixNano()
 	b.mime = strings.ToLower(codec.MimeType)
+	b.rtpParameters = params
 	for _, codecParameter := range params.Codecs {
 		if strings.EqualFold(codecParameter.MimeType, codec.MimeType) {
 			b.payloadType = uint8(codecParameter.PayloadType)
@@ -226,23 +232,23 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 		b.payloadType = uint8(params.Codecs[0].PayloadType)
 	}
 
+	// find RTX payload type
+	for _, codec := range params.Codecs {
+		if strings.EqualFold(codec.MimeType, "video/rtx") && strings.Contains(codec.SDPFmtpLine, fmt.Sprintf("apt=%d", b.payloadType)) {
+			b.rtxPayloadType = uint8(codec.PayloadType)
+			break
+		}
+	}
+
 	for _, ext := range params.HeaderExtensions {
 		switch ext.URI {
 		case dd.ExtensionURI:
-			if IsSvcCodec(codec.MimeType) || strings.EqualFold(codec.MimeType, webrtc.MimeTypeVP8) {
-				if b.ddExtID != 0 {
-					b.logger.Warnw("multiple dependency descriptor extensions found", nil, "id", ext.ID, "previous", b.ddExtID)
-					continue
-				}
-				b.ddExtID = uint8(ext.ID)
-				frc := NewFrameRateCalculatorDD(b.clockRate, b.logger)
-				for i := range b.frameRateCalculator {
-					b.frameRateCalculator[i] = frc.GetFrameRateCalculatorForSpatial(int32(i))
-				}
-				b.ddParser = NewDependencyDescriptorParser(b.ddExtID, b.logger, func(spatial, temporal int32) {
-					frc.SetMaxLayer(spatial, temporal)
-				})
+			if b.ddExtID != 0 {
+				b.logger.Warnw("multiple dependency descriptor extensions found", nil, "id", ext.ID, "previous", b.ddExtID)
+				continue
 			}
+			b.ddExtID = uint8(ext.ID)
+			b.createDDParserAndFrameRateCalculator(codec.MimeType)
 
 		case sdp.AudioLevelURI:
 			b.audioLevelExtID = uint8(ext.ID)
@@ -262,16 +268,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 		b.codecType = webrtc.RTPCodecTypeVideo
 		b.bucket = bucket.NewBucket[uint64](InitPacketBufferSizeVideo)
 		if b.frameRateCalculator[0] == nil {
-			if strings.EqualFold(codec.MimeType, webrtc.MimeTypeVP8) {
-				b.frameRateCalculator[0] = NewFrameRateCalculatorVP8(b.clockRate, b.logger)
-			}
-
-			if strings.EqualFold(codec.MimeType, webrtc.MimeTypeVP9) {
-				frc := NewFrameRateCalculatorVP9(b.clockRate, b.logger)
-				for i := range b.frameRateCalculator {
-					b.frameRateCalculator[i] = frc.GetFrameRateCalculatorForSpatial(int32(i))
-				}
-			}
+			b.createFrameRateCalculator(codec.MimeType)
 		}
 		if bitrates > 0 {
 			pps := bitrates / 8 / 1200
@@ -310,6 +307,40 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 	b.bound = true
 }
 
+func (b *Buffer) OnCodecChange(fn func(webrtc.RTPCodecParameters)) {
+	b.Lock()
+	b.onCodecChange = fn
+	b.Unlock()
+}
+
+func (b *Buffer) createDDParserAndFrameRateCalculator(mime string) {
+	if IsSvcCodec(mime) || strings.EqualFold(mime, webrtc.MimeTypeVP8) {
+		frc := NewFrameRateCalculatorDD(b.clockRate, b.logger)
+		for i := range b.frameRateCalculator {
+			b.frameRateCalculator[i] = frc.GetFrameRateCalculatorForSpatial(int32(i))
+		}
+		b.ddParser = NewDependencyDescriptorParser(b.ddExtID, b.logger, func(spatial, temporal int32) {
+			frc.SetMaxLayer(spatial, temporal)
+		})
+	}
+}
+
+func (b *Buffer) createFrameRateCalculator(mime string) {
+	switch {
+	case strings.EqualFold(mime, webrtc.MimeTypeVP8):
+		b.frameRateCalculator[0] = NewFrameRateCalculatorVP8(b.clockRate, b.logger)
+
+	case strings.EqualFold(mime, webrtc.MimeTypeVP9):
+		frc := NewFrameRateCalculatorVP9(b.clockRate, b.logger)
+		for i := range b.frameRateCalculator {
+			b.frameRateCalculator[i] = frc.GetFrameRateCalculatorForSpatial(int32(i))
+		}
+
+	case strings.EqualFold(mime, webrtc.MimeTypeH265):
+		b.frameRateCalculator[0] = NewFrameRateCalculatorH26x(b.clockRate, b.logger)
+	}
+}
+
 // Write adds an RTP Packet, ordering is not guaranteed, newer packets may arrive later
 func (b *Buffer) Write(pkt []byte) (n int, err error) {
 	var rtpPacket rtp.Packet
@@ -364,7 +395,6 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) {
 		return
 	}
 
-	b.payloadType = rtpPacket.PayloadType
 	b.calc(pkt, &rtpPacket, now, false)
 	b.Unlock()
 	b.readCond.Broadcast()
@@ -397,6 +427,11 @@ func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) (n int, err err
 		return
 	}
 
+	if rtxPkt.PayloadType != b.rtxPayloadType {
+		b.logger.Debugw("unexpected rtx payload type", "expected", b.rtxPayloadType, "actual", rtxPkt.PayloadType)
+		return
+	}
+
 	if b.rtxPktBuf == nil {
 		b.rtxPktBuf = make([]byte, bucket.MaxPktSize)
 	}
@@ -593,6 +628,10 @@ func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, i
 		return
 	}
 
+	if !flowState.IsOutOfOrder && rtpPacket.PayloadType != b.payloadType && b.codecType == webrtc.RTPCodecTypeVideo {
+		b.handleCodecChange(rtpPacket.PayloadType)
+	}
+
 	// add to RTX buffer using sequence number after accounting for dropped padding only packets
 	snAdjustment, err := b.snRangeMap.GetValue(flowState.ExtSequenceNumber)
 	if err != nil {
@@ -712,6 +751,55 @@ func (b *Buffer) doFpsCalc(ep *ExtPacket) {
 	}
 }
 
+func (b *Buffer) handleCodecChange(newPT uint8) {
+	var (
+		codecFound, rtxFound bool
+		rtxPt                uint8
+		newCodec             webrtc.RTPCodecParameters
+	)
+	for _, codec := range b.rtpParameters.Codecs {
+		if !codecFound && uint8(codec.PayloadType) == newPT {
+			newCodec = codec
+			codecFound = true
+		}
+
+		if strings.EqualFold(codec.MimeType, "video/rtx") && strings.Contains(codec.SDPFmtpLine, fmt.Sprintf("apt=%d", newPT)) {
+			rtxFound = true
+			rtxPt = uint8(codec.PayloadType)
+		}
+
+		if codecFound && rtxFound {
+			break
+		}
+	}
+	if !codecFound {
+		b.logger.Errorw("could not find codec for new payload type", nil, "pt", newPT, "rtpParameters", b.rtpParameters)
+		return
+	}
+	b.logger.Infow("codec changed",
+		"oldPayload", b.payloadType, "newPayload", newPT,
+		"oldRtxPayload", b.rtxPayloadType, "newRtxPayload", rtxPt,
+		"oldMime", b.mime, "newMime", newCodec.MimeType)
+	b.payloadType = newPT
+	b.rtxPayloadType = rtxPt
+	b.mime = strings.ToLower(newCodec.MimeType)
+	b.frameRateCalculated = false
+
+	if b.ddExtID != 0 {
+		b.createDDParserAndFrameRateCalculator(b.mime)
+	}
+
+	if b.frameRateCalculator[0] == nil {
+		b.createFrameRateCalculator(b.mime)
+	}
+
+	b.bucket.ResyncOnNextPacket()
+
+	if f := b.onCodecChange; f != nil {
+		go f(newCodec)
+	}
+}
+
 func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime int64) rtpstats.RTPFlowState {
 	flowState := b.rtpStats.Update(
 		arrivalTime,
@@ -825,6 +913,19 @@ func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64, flowStat
 
 	case utils.MimeTypeAV1:
 		ep.KeyFrame = IsAV1KeyFrame(rtpPacket.Payload)
+
+	case utils.MimeTypeH265:
+		if ep.DependencyDescriptor == nil {
+			if len(rtpPacket.Payload) < 2 {
+				b.logger.Warnw("invalid H265 packet", nil)
+				return nil
+			}
+			ep.VideoLayer = VideoLayer{
+				Temporal: int32(rtpPacket.Payload[1]&0x07) - 1,
+			}
+			ep.Spatial = InvalidLayerSpatial
+		}
+		ep.KeyFrame = IsH265KeyFrame(rtpPacket.Payload)
 	}
 
 	if ep.KeyFrame {

pkg/sfu/buffer/buffer_test.go

@@ -28,6 +28,17 @@ import (
 	"github.com/livekit/mediatransportutil/pkg/nack"
 )
 
+var h265Codec = webrtc.RTPCodecParameters{
+	RTPCodecCapability: webrtc.RTPCodecCapability{
+		MimeType:  "video/h265",
+		ClockRate: 90000,
+		RTCPFeedback: []webrtc.RTCPFeedback{{
+			Type: "nack",
+		}},
+	},
+	PayloadType: 116,
+}
+
 var vp8Codec = webrtc.RTPCodecParameters{
 	RTPCodecCapability: webrtc.RTPCodecCapability{
 		MimeType:  "video/vp8",
@@ -314,6 +325,105 @@ func TestFractionLostReport(t *testing.T) {
 	wg.Wait()
 }
 
+func TestCodecChange(t *testing.T) {
+	// codec change before bind
+	buff := NewBuffer(123, 1, 1)
+	require.NotNil(t, buff)
+	changedCodec := make(chan webrtc.RTPCodecParameters, 1)
+	buff.OnCodecChange(func(rp webrtc.RTPCodecParameters) {
+		select {
+		case changedCodec <- rp:
+		default:
+			t.Fatalf("codec change not consumed")
+		}
+	})
+
+	h265Pkt := rtp.Packet{
+		Header: rtp.Header{
+			Version:        2,
+			PayloadType:    116,
+			SequenceNumber: 1,
+			Timestamp:      1,
+			SSRC:           123,
+		},
+		Payload: []byte{0xff, 0xff, 0xff, 0xfd, 0xb4, 0x9f, 0x94, 0x1},
+	}
+	buf, err := h265Pkt.Marshal()
+	require.NoError(t, err)
+	_, err = buff.Write(buf)
+	require.NoError(t, err)
+
+	select {
+	case <-changedCodec:
+		t.Fatalf("unexpected codec change")
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	buff.Bind(webrtc.RTPParameters{
+		HeaderExtensions: nil,
+		Codecs:           []webrtc.RTPCodecParameters{vp8Codec, h265Codec},
+	}, vp8Codec.RTPCodecCapability, 0)
+
+	select {
+	case c := <-changedCodec:
+		require.Equal(t, h265Codec, c)
+
+	case <-time.After(1 * time.Second):
+		t.Fatalf("expected codec change")
+	}
+
+	// codec change after bind
+	vp8Pkt := rtp.Packet{
+		Header: rtp.Header{
+			Version:        2,
+			PayloadType:    96,
+			SequenceNumber: 3,
+			Timestamp:      3,
+			SSRC:           123,
+		},
+		Payload: []byte{0xff, 0xff, 0xff, 0xfd, 0xb4, 0x9f, 0x94, 0x1},
+	}
+	buf, err = vp8Pkt.Marshal()
+	require.NoError(t, err)
+	_, err = buff.Write(buf)
+	require.NoError(t, err)
+
+	select {
+	case c := <-changedCodec:
+		require.Equal(t, vp8Codec, c)
+
+	case <-time.After(1 * time.Second):
+		t.Fatalf("expected codec change")
+	}
+
+	// out of order pkts can't cause codec change
+	h265Pkt.SequenceNumber = 2
+	h265Pkt.Timestamp = 2
+	buf, err = h265Pkt.Marshal()
+	require.NoError(t, err)
+	_, err = buff.Write(buf)
+	require.NoError(t, err)
+	select {
+	case <-changedCodec:
+		t.Fatalf("unexpected codec change")
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	// unknown codec should not cause change
+	h265Pkt.SequenceNumber = 4
+	h265Pkt.Timestamp = 4
+	h265Pkt.PayloadType = 117
+	buf, err = h265Pkt.Marshal()
+	require.NoError(t, err)
+	_, err = buff.Write(buf)
+	require.NoError(t, err)
+	select {
+	case <-changedCodec:
+		t.Fatalf("unexpected codec change")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
 func BenchmarkMemcpu(b *testing.B) {
 	buf := make([]byte, 1500*1500*10)
 	buf2 := make([]byte, 1500*1500*20)

pkg/sfu/buffer/fps.go

@@ -17,14 +17,16 @@ package buffer
 import (
 	"container/list"
 
-	"github.com/livekit/protocol/logger"
 	"github.com/pion/rtp/codecs"
+
+	"github.com/livekit/protocol/logger"
 )
 
 var minFramesForCalculation = [...]int{8, 15, 40, 60}
 
 type frameInfo struct {
-	seq       uint16
+	startSeq  uint16
+	endSeq    uint16
 	ts        uint32
 	fn        uint16
 	spatial   int32
@@ -79,7 +81,7 @@ func (f *frameRateCalculatorVPx) RecvPacket(ep *ExtPacket, fn uint16) bool {
 	}
 
 	if f.baseFrame == nil {
-		f.baseFrame = &frameInfo{seq: ep.Packet.SequenceNumber, ts: ep.Packet.Timestamp, fn: fn}
+		f.baseFrame = &frameInfo{ts: ep.Packet.Timestamp, fn: fn}
 		f.fnReceived[0] = f.baseFrame
 		f.firstFrames[temporal] = f.baseFrame
 		return false
@@ -102,7 +104,6 @@ func (f *frameRateCalculatorVPx) RecvPacket(ep *ExtPacket, fn uint16) bool {
 	}
 
 	fi := &frameInfo{
-		seq:      ep.Packet.SequenceNumber,
 		ts:       ep.Packet.Timestamp,
 		fn:       fn,
 		temporal: temporal,
@@ -373,7 +374,7 @@ func (f *FrameRateCalculatorDD) RecvPacket(ep *ExtPacket) bool {
 
 	fn := ep.DependencyDescriptor.Descriptor.FrameNumber
 	if f.baseFrame == nil {
-		f.baseFrame = &frameInfo{seq: ep.Packet.SequenceNumber, ts: ep.Packet.Timestamp, fn: fn}
+		f.baseFrame = &frameInfo{ts: ep.Packet.Timestamp, fn: fn}
 		f.fnReceived[0] = f.baseFrame
 		f.firstFrames[spatial][temporal] = f.baseFrame
 		f.secondFrames[spatial][temporal] = f.baseFrame
@@ -406,7 +407,6 @@ func (f *FrameRateCalculatorDD) RecvPacket(ep *ExtPacket) bool {
 	}
 
 	fi := &frameInfo{
-		seq:       ep.Packet.SequenceNumber,
 		ts:        ep.Packet.Timestamp,
 		fn:        fn,
 		temporal:  temporal,
@@ -438,12 +438,13 @@ func (f *FrameRateCalculatorDD) RecvPacket(ep *ExtPacket) bool {
 			switch {
 			case val == dependFrame:
 				break insertFrame
-			case val < dependFrame:
+			case sn16LT(val, dependFrame):
 				chain.InsertAfter(dependFrame, e)
 				break insertFrame
 			default:
 				if e == chain.Front() {
 					chain.PushFront(dependFrame)
+					break insertFrame
 				}
 			}
 		}
@@ -570,3 +571,162 @@ func (f *FrameRateCalculatorForDDLayer) GetFrameRate() []float32 {
 }
 
 // -----------------------------------------------
+
+type FrameRateCalculatorH26x struct {
+	frameRates [DefaultMaxLayerTemporal + 1]float32
+	clockRate  uint32
+	logger     logger.Logger
+	fnReceived *list.List
+	baseFrame  *frameInfo
+	completed  bool
+}
+
+func NewFrameRateCalculatorH26x(clockRate uint32, logger logger.Logger) *FrameRateCalculatorH26x {
+	return &FrameRateCalculatorH26x{
+		clockRate: clockRate,
+		logger:    logger,
+	}
+}
+
+func (f *FrameRateCalculatorH26x) Completed() bool {
+	return f.completed
+}
+
+func (f *FrameRateCalculatorH26x) RecvPacket(ep *ExtPacket) bool {
+	if f.completed {
+		return true
+	}
+
+	if ep.Temporal >= int32(len(f.frameRates)) {
+		f.logger.Warnw("invalid temporal layer", nil, "temporal", ep.Temporal)
+		return false
+	}
+
+	temporal := ep.Temporal
+	if temporal < 0 {
+		temporal = 0
+	}
+
+	if f.baseFrame == nil {
+		f.baseFrame = &frameInfo{
+			startSeq: ep.Packet.SequenceNumber,
+			endSeq:   ep.Packet.SequenceNumber,
+			ts:       ep.Packet.Timestamp,
+			temporal: temporal,
+		}
+		f.fnReceived = list.New()
+		f.fnReceived.PushBack(f.baseFrame)
+		return false
+	}
+
+	if sn16LTOrEqual(ep.Packet.SequenceNumber, f.baseFrame.startSeq) {
+		return false
+	}
+
+insertFrame:
+	for e := f.fnReceived.Back(); e != nil; e = e.Prev() {
+		frame := e.Value.(*frameInfo)
+		switch {
+		case frame.ts == ep.Packet.Timestamp:
+			if sn16LT(frame.endSeq, ep.Packet.SequenceNumber) {
+				frame.endSeq = ep.Packet.SequenceNumber
+			}
+			if sn16LT(ep.Packet.SequenceNumber, frame.startSeq) {
+				frame.startSeq = ep.Packet.SequenceNumber
+			}
+			break insertFrame
+		case sn32LT(frame.ts, ep.Packet.Timestamp):
+			f.fnReceived.InsertAfter(&frameInfo{
+				startSeq: ep.Packet.SequenceNumber,
+				endSeq:   ep.Packet.SequenceNumber,
+				ts:       ep.Packet.Timestamp,
+				temporal: temporal,
+			}, e)
+			break insertFrame
+		default:
+			if e == f.fnReceived.Front() {
+				f.fnReceived.PushFront(&frameInfo{
+					startSeq: ep.Packet.SequenceNumber,
+					endSeq:   ep.Packet.SequenceNumber,
+					ts:       ep.Packet.Timestamp,
+					temporal: temporal,
+				})
+				break insertFrame
+			}
+		}
+	}
+
+	return f.calc()
+}
+
+func (f *FrameRateCalculatorH26x) calc() bool {
+	frameCounts := make([]int, DefaultMaxLayerTemporal+1)
+	var totalFrameCount int
+	var tsDuration int
+	cur := f.fnReceived.Front()
+	for {
+		next := cur.Next()
+		if next == nil {
+			break
+		}
+		ff := cur.Value.(*frameInfo)
+		nf := next.Value.(*frameInfo)
+		if nf.startSeq-ff.endSeq == 1 {
+			totalFrameCount++
+			tsDuration += int(nf.ts - ff.ts)
+			for i := int(nf.temporal); i < len(frameCounts); i++ {
+				frameCounts[i]++
+			}
+		} else {
+			// reset to find continuous frames
+			totalFrameCount = 0
+			for i := range frameCounts {
+				frameCounts[i] = 0
+			}
+			tsDuration = 0
+		}
+
+		// received enough continuous frames, calculate fps
+		if totalFrameCount >= minFramesForCalculation[DefaultMaxLayerTemporal] {
+			for currentTemporal := int32(0); currentTemporal <= DefaultMaxLayerTemporal; currentTemporal++ {
+				count := frameCounts[currentTemporal]
+				if currentTemporal > 0 && count == frameCounts[currentTemporal-1] {
+					// no frames for this temporal layer
+					f.frameRates[currentTemporal] = 0
+				} else {
+					f.frameRates[currentTemporal] = float32(f.clockRate) / float32(tsDuration) * float32(count)
+				}
+			}
+			f.logger.Debugw("fps changed", "fps", f.GetFrameRate())
+			f.completed = true
+			f.reset()
+			return true
+		}
+
+		cur = next
+	}
+
+	return false
+}
+
+func (f *FrameRateCalculatorH26x) reset() {
+	f.fnReceived.Init()
+	f.baseFrame = nil
+}
+
+func (f *FrameRateCalculatorH26x) GetFrameRate() []float32 {
+	return f.frameRates[:]
+}
+
+// -----------------------------------------------
+func sn16LT(a, b uint16) bool {
+	return a-b > 0x8000
+}
+
+func sn16LTOrEqual(a, b uint16) bool {
+	return a == b || a-b > 0x8000
+}
+
+func sn32LT(a, b uint32) bool {
+	return a-b > 0x80000000
+}

pkg/sfu/buffer/fps_test.go

@@ -55,10 +55,16 @@ func (f *testFrameInfo) toDD() *ExtPacket {
 		},
 		VideoLayer: VideoLayer{Spatial: int32(f.spatial), Temporal: int32(f.temporal)},
 	}
-
 }
 
-func createFrames(startFrameNumber uint16, startTs uint32, totalFramesPerSpatial int, fps [][]float32, spatialDependency bool) [][]*testFrameInfo {
+func (f *testFrameInfo) toH26x() *ExtPacket {
+	return &ExtPacket{
+		Packet:     &rtp.Packet{Header: f.header},
+		VideoLayer: VideoLayer{Spatial: InvalidLayerSpatial, Temporal: int32(f.temporal)},
+	}
+}
+
+func createFrames(startFrameNumber uint16, startTs uint32, startSeq uint16, totalFramesPerSpatial int, fps [][]float32, spatialDependency bool) [][]*testFrameInfo {
 	spatials := len(fps)
 	temporals := len(fps[0])
 	frames := make([][]*testFrameInfo, spatials)
@@ -85,7 +91,7 @@ func createFrames(startFrameNumber uint16, startTs uint32, totalFramesPerSpatial
 	for i := 0; i < totalFramesPerSpatial; i++ {
 		for s := 0; s < spatials; s++ {
 			frame := &testFrameInfo{
-				header:      rtp.Header{Timestamp: currentTs[s]},
+				header:      rtp.Header{Timestamp: currentTs[s], SequenceNumber: startSeq},
 				framenumber: fn,
 				spatial:     s,
 			}
@@ -101,6 +107,7 @@ func createFrames(startFrameNumber uint16, startTs uint32, totalFramesPerSpatial
 			currentTs[s] += tsStep[s][temporals-1]
 			frames[s] = append(frames[s], frame)
 			fn++
+			startSeq++
 
 			for fidx := len(frames[s]) - 1; fidx >= 0; fidx-- {
 				cf := frames[s][fidx]
@@ -135,6 +142,7 @@ func verifyFps(t *testing.T, expect, got []float32) {
 
 type testcase struct {
 	startTs           uint32
+	startSeq          uint16
 	startFrameNumber  uint16
 	fps               [][]float32
 	spatialDependency bool
@@ -167,7 +175,7 @@ func TestFpsVP8(t *testing.T) {
 			vp8calcs := make([]*FrameRateCalculatorVP8, len(fps))
 			for i := range vp8calcs {
 				vp8calcs[i] = NewFrameRateCalculatorVP8(90000, logger.GetLogger())
-				frames = append(frames, createFrames(c.startFrameNumber, c.startTs, 200, [][]float32{fps[i]}, false)[0])
+				frames = append(frames, createFrames(c.startFrameNumber, c.startTs, 10, 200, [][]float32{fps[i]}, false)[0])
 			}
 
 			var frameratesGot bool
@@ -198,7 +206,7 @@ func TestFpsVP8(t *testing.T) {
 		vp8calcs := make([]*FrameRateCalculatorVP8, len(fps))
 		for i := range vp8calcs {
 			vp8calcs[i] = NewFrameRateCalculatorVP8(90000, logger.GetLogger())
-			frames = append(frames, createFrames(100, 12345678, 300, [][]float32{fps[i]}, false)[0])
+			frames = append(frames, createFrames(100, 12345678, 10, 300, [][]float32{fps[i]}, false)[0])
 			for j := 5; j < 130; j++ {
 				if j%2 == 0 {
 					frames[i][j] = frames[i][j-1]
@@ -255,7 +263,7 @@ func TestFpsDD(t *testing.T) {
 		testCase := c
 		t.Run(name, func(t *testing.T) {
 			fps := testCase.fps
-			frames := createFrames(c.startFrameNumber, c.startTs, 500, fps, testCase.spatialDependency)
+			frames := createFrames(c.startFrameNumber, c.startTs, 10, 500, fps, testCase.spatialDependency)
 			ddcalc := NewFrameRateCalculatorDD(90000, logger.GetLogger())
 			ddcalc.SetMaxLayer(int32(len(fps)-1), int32(len(fps[0])-1))
 			ddcalcs := make([]FrameRateCalculator, len(fps))
@@ -288,7 +296,7 @@ func TestFpsDD(t *testing.T) {
 
 	t.Run("packet lost and duplicate", func(t *testing.T) {
 		fps := [][]float32{{7.5, 15, 30}, {7.5, 15, 30}, {7.5, 15, 30}}
-		frames := createFrames(100, 12345678, 500, fps, true)
+		frames := createFrames(100, 12345678, 10, 500, fps, true)
 		ddcalc := NewFrameRateCalculatorDD(90000, logger.GetLogger())
 		ddcalc.SetMaxLayer(int32(len(fps)-1), int32(len(fps[0])-1))
 		ddcalcs := make([]FrameRateCalculator, len(fps))
@@ -322,5 +330,108 @@ func TestFpsDD(t *testing.T) {
 			verifyFps(t, fpsExpected, fpsGot[:len(fpsExpected)])
 		}
 	})
-
+}
+
+func TestFpsH26x(t *testing.T) {
+	cases := map[string]testcase{
+		"normal": {
+			startTs:          12345678,
+			startSeq:         100,
+			startFrameNumber: 100,
+			fps:              [][]float32{{5, 10, 15}, {5, 10, 15}, {7.5, 15, 30}},
+		},
+		"frame number and timestamp wrap": {
+			startTs:          (uint32(1) << 31) - 10,
+			startSeq:         (uint16(1) << 15) - 10,
+			startFrameNumber: (uint16(1) << 15) - 10,
+			fps:              [][]float32{{5, 10, 15}, {5, 10, 15}, {7.5, 15, 30}},
+		},
+		"2 temporal layers": {
+			startTs:          12345678,
+			startFrameNumber: 100,
+			fps:              [][]float32{{7.5, 15}, {7.5, 15}, {15, 30}},
+		},
+	}
+
+	for name, c := range cases {
+		testCase := c
+		t.Run(name, func(t *testing.T) {
+			fps := testCase.fps
+			frames := make([][]*testFrameInfo, 0)
+			h26xcalcs := make([]*FrameRateCalculatorH26x, len(fps))
+			for i := range h26xcalcs {
+				h26xcalcs[i] = NewFrameRateCalculatorH26x(90000, logger.GetLogger())
+				frames = append(frames, createFrames(c.startFrameNumber, c.startTs, c.startSeq, 200, [][]float32{fps[i]}, false)[0])
+			}
+
+			var frameratesGot bool
+			for s, fs := range frames {
+				for _, f := range fs {
+					if h26xcalcs[s].RecvPacket(f.toH26x()) {
+						frameratesGot = true
+						for _, calc := range h26xcalcs {
+							if !calc.Completed() {
+								frameratesGot = false
+								break
+							}
+						}
+					}
+				}
+			}
+			require.True(t, frameratesGot)
+			for i, calc := range h26xcalcs {
+				fpsExpected := fps[i]
+				fpsGot := calc.GetFrameRate()
+				verifyFps(t, fpsExpected, fpsGot[:len(fpsExpected)])
+			}
+		})
+	}
+
+	t.Run("packet lost and duplicate", func(t *testing.T) {
+		fps := [][]float32{{7.5, 15, 30}, {7.5, 15, 30}, {7.5, 15, 30}}
+		frames := make([][]*testFrameInfo, 0, len(fps))
+		h26xcalcs := make([]FrameRateCalculator, len(fps))
+		for i := range fps {
+			frames = append(frames, createFrames(100, 12345678, 10, 500, [][]float32{fps[i]}, false)[0])
+			h26xcalcs[i] = NewFrameRateCalculatorH26x(90000, logger.GetLogger())
+			for j := 5; j < 130; j++ {
+				if j%2 == 0 {
+					frames[i][j] = frames[i][j-1]
+				}
+			}
+			for j := 130; j < 230; j++ {
+				if j%3 == 0 {
+					frames[i][j] = nil
+				}
+			}
+			for j := 230; j < 330; j++ {
+				if j%2 == 0 {
+					frames[i][j], frames[i][j-1] = frames[i][j-1], frames[i][j]
+				}
+			}
+		}
+		var frameratesGot bool
+		for s, fs := range frames {
+			for _, f := range fs {
+				if f == nil {
+					continue
+				}
+				if h26xcalcs[s].RecvPacket(f.toH26x()) {
+					frameratesGot = true
+					for _, calc := range h26xcalcs {
+						if !calc.Completed() {
+							frameratesGot = false
+							break
+						}
+					}
+				}
+			}
+		}
+		require.True(t, frameratesGot)
+		for i, calc := range h26xcalcs {
+			fpsExpected := fps[i]
+			fpsGot := calc.GetFrameRate()
+			verifyFps(t, fpsExpected, fpsGot[:len(fpsExpected)])
+		}
+	})
 }

pkg/sfu/buffer/helpers.go

@@ -18,8 +18,9 @@ import (
 	"encoding/binary"
 	"errors"
 
-	"github.com/livekit/protocol/logger"
 	"github.com/pion/rtp/codecs"
+
+	"github.com/livekit/protocol/logger"
 )
 
 var (
@@ -415,4 +416,40 @@ func IsAV1KeyFrame(payload []byte) bool {
 	}
 }
 
+func IsH265KeyFrame(payload []byte) (kf bool) {
+	if len(payload) < 2 {
+		return false
+	}
+	naluType := (payload[0] & 0x7E) >> 1
+	switch {
+	case naluType == 33 || naluType == 34:
+		return true
+	case naluType == 48: // AP
+		idx := 2
+		for idx < len(payload)-2 {
+			// TODO: check the DONL field (controled by sprop-max-don-diff)
+			size := binary.BigEndian.Uint16(payload[idx:])
+			idx += 2
+			if idx >= len(payload) {
+				return false
+			}
+			naluType = (payload[idx] & 0x7E) >> 1
+			if naluType == 33 || naluType == 34 {
+				return true
+			}
+			idx += int(size)
+		}
+		return false
+
+	case naluType == 49: // FU
+		if len(payload) < 3 {
+			return false
+		}
+		naluType = (payload[2] & 0x7E) >> 1
+		return naluType == 33 || naluType == 34
+	default:
+		return false
+	}
+}
+
 // -------------------------------------