sfu: resync dependency descriptor frame numbers on unmute (#5094)

On mute the forwarder resyncs sequence numbers, timestamps and VP8 picture
IDs, but DDForwarder kept its frame number offset. libwebrtc counts DD
frames across all simulcast layers, so after a long mute the frame number
sent on unmute jumped by more than half the 16 bit range. The subscriber
took every frame as older than the last decoded one and dropped it, a
black tile until the numbers wrapped past it again.

Resync DDForwarder with the forwarder, and move the numbers to just after
the last one sent when they jump too far ahead.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

pkg/sfu/forwarder.go

@@ -1787,6 +1787,9 @@ func (f *Forwarder) Resync() {
 func (f *Forwarder) resyncLocked() {
 	f.vls.SetCurrent(buffer.InvalidLayer)
 	f.lastSSRC = 0
+	if f.ddForwarder != nil {
+		f.ddForwarder.Resync()
+	}
 	if f.pubMuted {
 		f.resumeBehindThreshold = ResumeBehindThresholdSeconds
 	}

pkg/sfu/videolayerselector/ddforwarder.go

@@ -30,10 +30,17 @@ import (
 // frame numbers of a new layer are close to those of the old one. The numbers sent to the
 // subscriber must still never go back: a key frame older than the last structure the
 // subscriber received is dropped. So whenever the stream changes, a layer switch, a stream
-// restart or a new forwarder, the numbers are moved up past the last one sent if needed.
+// restart, a resync or a new forwarder, the numbers are moved up past the last one sent if
+// needed. They must not jump too far ahead either: the subscriber unwraps the 16 bit frame
+// number, so a jump of half its range or more reads as a jump back. The publisher keeps
+// counting while a muted subscriber gets nothing, so a long mute would cause such a jump.
 // Otherwise they pass through with the current offset, so frame diffs and chain diffs stay
 // valid. A migrated subscriber continues with the last number sent, which is part of the
 // migrated forwarder state.
+// maxFrameNumJump is the largest jump in the frame numbers sent on a stream change, well below
+// half the range of the 16 bit frame number.
+const maxFrameNumJump = 1 << 14
+
 type DDForwarder struct {
 	logger logger.Logger
 
@@ -159,8 +166,18 @@ func (d *DDForwarder) Forward(extPkt *buffer.ExtPacket, temporal int32, result *
 }
 
 // rebase moves the frame numbers up when a new stream would go back past the last one sent
+// Resync starts the stream again on the next key frame with a structure, like a new forwarder,
+// keeping the last number sent. The forwarder resyncs on a mute, for example.
+func (d *DDForwarder) Resync() {
+	d.initialized = false
+}
+
 func (d *DDForwarder) rebase(extFrameNum uint64) {
-	if d.hasLast && extFrameNum+d.frameNumOffset <= d.extLastFrameNum {
+	if !d.hasLast {
+		return
+	}
+	extOutFrameNum := extFrameNum + d.frameNumOffset
+	if extOutFrameNum <= d.extLastFrameNum || extOutFrameNum-d.extLastFrameNum > maxFrameNumJump {
 		d.frameNumOffset = d.extLastFrameNum + 1 - extFrameNum
 	}
 }

pkg/sfu/videolayerselector/ddforwarder_test.go

@@ -183,6 +183,24 @@ func TestDDForwarderRestartOnDeltaFrame(t *testing.T) {
 	require.Equal(t, uint16(1004), d.forward(f, d.packet(1, 5, false, 1), 0).FrameNumber)
 }
 
+func TestDDForwarderResync(t *testing.T) {
+	d := newDDForwarderTest(t)
+	f := NewDDForwarder(logger.GetLogger())
+	require.Equal(t, uint16(1000), d.forward(f, d.packet(1, 1000, true, 0), 0).FrameNumber)
+	require.Equal(t, uint16(1001), d.forward(f, d.packet(1, 1001, false, 0), 0).FrameNumber)
+
+	// a long mute: the publisher counted 40000 frames while nothing was forwarded, the
+	// subscriber would take a jump of more than half the 16 bit range as a jump back
+	f.Resync()
+	require.Nil(t, d.forward(f, d.packet(1, 41001, false, 0), 0))
+	require.Equal(t, uint16(1002), d.forward(f, d.packet(1, 41002, true, 0), 0).FrameNumber)
+	require.Equal(t, uint16(1003), d.forward(f, d.packet(1, 41003, false, 0), 0).FrameNumber)
+
+	// a short mute keeps the current offset, the gap stays
+	f.Resync()
+	require.Equal(t, uint16(1103), d.forward(f, d.packet(1, 41103, true, 0), 0).FrameNumber)
+}
+
 func TestDDForwarderPublisherActiveDecodeTargets(t *testing.T) {
 	d := newDDForwarderTest(t)
 	f := NewDDForwarder(logger.GetLogger())