handle frame number wrap back in svc (#3885)

* handle frame number wrap back in svc

* Add Slack Notifier

* check nil dd ext

* log format

.github/workflows/slack-notifier.yaml

@@ -0,0 +1,25 @@
+name: PR Slack Notifier
+
+on:
+  pull_request:
+    types: [review_requested, reopened, closed]
+  pull_request_review:
+    types: [submitted]
+
+permissions:
+  contents: read
+  pull-requests: write
+  issues: write
+
+concurrency:
+  group: pr-slack-${{ github.event.pull_request.number }}-${{ github.workflow }}
+  cancel-in-progress: false
+
+jobs:
+  notify-devs:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: livekit/slack-notifier-action@main
+        with:
+          config_json: ${{ secrets.SLACK_NOTIFY_CONFIG_JSON }}
+          slack_token: ${{ secrets.SLACK_PR_NOTIFIER_TOKEN }}

pkg/sfu/buffer/buffer.go

@@ -338,7 +338,7 @@ func (b *Buffer) createDDParserAndFrameRateCalculator() {
 		}
 		b.ddParser = NewDependencyDescriptorParser(b.ddExtID, b.logger, func(spatial, temporal int32) {
 			frc.SetMaxLayer(spatial, temporal)
-		})
+		}, false)
 	}
 }
 

pkg/sfu/buffer/dependencydescriptorparser.go

@@ -17,6 +17,7 @@ package buffer
 import (
 	"fmt"
 	"sort"
+	"time"
 
 	"github.com/pion/rtp"
 	"go.uber.org/atomic"
@@ -27,6 +28,14 @@ import (
 	"github.com/livekit/protocol/logger"
 )
 
+const (
+	ddRestartThreshold = 30 * time.Second
+
+	// frame integrity check 2 seconds for L3T3 30fps video
+	integrityCheckFrame = 180
+	integrityCheckPkt   = 1024
+)
+
 var (
 	ErrFrameEarlierThanKeyFrame            = fmt.Errorf("frame is earlier than current keyframe")
 	ErrDDStructureAttachedToNonFirstPacket = fmt.Errorf("dependency descriptor structure is attached to non-first packet of a frame")
@@ -48,16 +57,22 @@ type DependencyDescriptorParser struct {
 	frameChecker              *FrameIntegrityChecker
 
 	ddNotFoundCount atomic.Uint32
+
+	// restart detection
+	restartGeneration int
+	enableRestart     bool
+	lastPacketAt      time.Time
 }
 
-func NewDependencyDescriptorParser(ddExtID uint8, logger logger.Logger, onMaxLayerChanged func(int32, int32)) *DependencyDescriptorParser {
+func NewDependencyDescriptorParser(ddExtID uint8, logger logger.Logger, onMaxLayerChanged func(int32, int32), enableRestart bool) *DependencyDescriptorParser {
 	return &DependencyDescriptorParser{
 		ddExtID:           ddExtID,
 		logger:            logger,
 		onMaxLayerChanged: onMaxLayerChanged,
 		seqWrapAround:     utils.NewWrapAround[uint16, uint64](utils.WrapAroundParams{IsRestartAllowed: false}),
 		frameWrapAround:   utils.NewWrapAround[uint16, uint64](utils.WrapAroundParams{IsRestartAllowed: false}),
-		frameChecker:      NewFrameIntegrityChecker(180, 1024), // 2seconds for L3T3 30fps video
+		frameChecker:      NewFrameIntegrityChecker(integrityCheckFrame, integrityCheckPkt),
+		enableRestart:     enableRestart,
 	}
 }
 
@@ -71,6 +86,10 @@ type ExtDependencyDescriptor struct {
 	ExtFrameNum                uint64
 	// the frame number of the keyframe which the current frame depends on
 	ExtKeyFrameNum uint64
+
+	// increase when the stream restarts, clear and reinitialize all dd state includes
+	// attached structure, frame chain, decode target.
+	RestartGeneration int
 }
 
 func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescriptor, VideoLayer, error) {
@@ -84,6 +103,16 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr
 		return nil, videoLayer, ErrDDExtentionNotFound
 	}
 
+	var restart bool
+	if r.enableRestart {
+		if !r.lastPacketAt.IsZero() && time.Since(r.lastPacketAt) > ddRestartThreshold {
+			r.restart()
+			restart = true
+			r.logger.Debugw("dependency descriptor parser restart stream", "generation", r.restartGeneration)
+		}
+		r.lastPacketAt = time.Now()
+	}
+
 	var ddVal dd.DependencyDescriptor
 	ext := &dd.DependencyDescriptorExtension{
 		Descriptor: &ddVal,
@@ -103,7 +132,8 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr
 		videoLayer.Spatial, videoLayer.Temporal = int32(ddVal.FrameDependencies.SpatialId), int32(ddVal.FrameDependencies.TemporalId)
 	}
 
-	unwrapped := r.frameWrapAround.Update(ddVal.FrameNumber)
+	// assume the packet is in-order when stream restarting
+	unwrapped := r.frameWrapAround.UpdateWithOrderKnown(ddVal.FrameNumber, restart)
 	extFN := unwrapped.ExtendedVal
 
 	if extFN < r.structureExtFrameNum {
@@ -114,9 +144,10 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr
 	r.frameChecker.AddPacket(extSeq, extFN, &ddVal)
 
 	extDD := &ExtDependencyDescriptor{
-		Descriptor:  &ddVal,
-		ExtFrameNum: extFN,
-		Integrity:   r.frameChecker.FrameIntegrity(extFN),
+		Descriptor:        &ddVal,
+		ExtFrameNum:       extFN,
+		Integrity:         r.frameChecker.FrameIntegrity(extFN),
+		RestartGeneration: r.restartGeneration,
 	}
 
 	if ddVal.AttachedStructure != nil {
@@ -167,6 +198,16 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr
 	return extDD, videoLayer, nil
 }
 
+func (r *DependencyDescriptorParser) restart() {
+	r.frameChecker = NewFrameIntegrityChecker(integrityCheckFrame, integrityCheckPkt)
+	r.structure = nil
+	r.structureExtFrameNum = 0
+	r.activeDecodeTargetsExtSeq = 0
+	r.activeDecodeTargetsMask = 0
+	r.decodeTargets = r.decodeTargets[:0]
+	r.restartGeneration++
+}
+
 // ------------------------------------------------------------------------------
 
 type DependencyDescriptorDecodeTarget struct {

pkg/sfu/utils/wraparound.go

@@ -80,7 +80,7 @@ func (w *WrapAroundUpdateResult[ET]) MarshalLogObject(e zapcore.ObjectEncoder) e
 	return nil
 }
 
-func (w *WrapAround[T, ET]) Update(val T) (result WrapAroundUpdateResult[ET]) {
+func (w *WrapAround[T, ET]) UpdateWithOrderKnown(val T, orderKnown bool) (result WrapAroundUpdateResult[ET]) {
 	if !w.initialized {
 		result.PreExtendedHighest = ET(val) - 1
 		result.ExtendedVal = ET(val)
@@ -92,10 +92,12 @@ func (w *WrapAround[T, ET]) Update(val T) (result WrapAroundUpdateResult[ET]) {
 		return
 	}
 
-	gap := val - w.highest
-	if gap > T(w.fullRange>>1) {
-		// out-of-order
-		return w.maybeAdjustStart(val)
+	if !orderKnown {
+		gap := val - w.highest
+		if gap > T(w.fullRange>>1) {
+			// out-of-order
+			return w.maybeAdjustStart(val)
+		}
 	}
 
 	// in-order
@@ -111,6 +113,10 @@ func (w *WrapAround[T, ET]) Update(val T) (result WrapAroundUpdateResult[ET]) {
 	return
 }
 
+func (w *WrapAround[T, ET]) Update(val T) (result WrapAroundUpdateResult[ET]) {
+	return w.UpdateWithOrderKnown(val, false)
+}
+
 func (w *WrapAround[T, ET]) UndoUpdate(result WrapAroundUpdateResult[ET]) {
 	if !w.initialized || result.PreExtendedHighest >= result.ExtendedVal {
 		return

pkg/sfu/videolayerselector/dependencydescriptor.go

@@ -24,6 +24,11 @@ import (
 	"github.com/livekit/protocol/logger"
 )
 
+const (
+	decisionCacheMaxElements = 256
+	decisionCacheNackEntries = 80
+)
+
 type DependencyDescriptor struct {
 	*Base
 
@@ -40,12 +45,14 @@ type DependencyDescriptor struct {
 	decodeTargetsLock sync.RWMutex
 	decodeTargets     []*DecodeTarget
 	fnWrapper         FrameNumberWrapper
+
+	restartGeneration int
 }
 
 func NewDependencyDescriptor(logger logger.Logger) *DependencyDescriptor {
 	return &DependencyDescriptor{
 		Base:      NewBase(logger),
-		decisions: NewSelectorDecisionCache(256, 80),
+		decisions: NewSelectorDecisionCache(decisionCacheMaxElements, decisionCacheNackEntries),
 		fnWrapper: FrameNumberWrapper{logger: logger},
 	}
 }
@@ -75,6 +82,20 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r
 		return
 	}
 
+	if ddwdt.RestartGeneration > d.restartGeneration {
+		d.logger.Debugw("stream restarted",
+			"packet", ddwdt.RestartGeneration,
+			"current", d.restartGeneration,
+			"structureKeyFrame", d.extKeyFrameNum,
+			"efn", ddwdt.ExtFrameNum,
+			"lastEfn", d.fnWrapper.LastOrigin(),
+		)
+		d.restart(ddwdt.RestartGeneration)
+	} else if ddwdt.RestartGeneration < d.restartGeneration {
+		// must not happen
+		d.logger.Warnw("packet from old generation", nil, "packet", ddwdt.RestartGeneration, "current", d.restartGeneration)
+	}
+
 	dd := ddwdt.Descriptor
 
 	extFrameNum := ddwdt.ExtFrameNum
@@ -434,3 +455,9 @@ func (d *DependencyDescriptor) CheckSync() (locked bool, layer int32) {
 
 	return false, layer
 }
+
+func (d *DependencyDescriptor) restart(generation int) {
+	d.restartGeneration = generation
+	d.invalidateKeyFrame()
+	d.decisions = NewSelectorDecisionCache(decisionCacheMaxElements, decisionCacheNackEntries)
+}

pkg/sfu/videolayerselector/framenumberwrapper.go

@@ -52,3 +52,7 @@ func (f *FrameNumberWrapper) UpdateAndGet(new uint64, updateOffset bool) uint64
 	f.last = new
 	return new + f.offset
 }
+
+func (f *FrameNumberWrapper) LastOrigin() uint64 {
+	return f.last
+}