From d5f5bcb60cbb4bf359e1e61360388adb6a58e8fb Mon Sep 17 00:00:00 2001 From: Grant <14.gsmith.14@gmail.com> Date: Thu, 29 Oct 2026 00:22:36 -0500 Subject: [PATCH] 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 --- pkg/sfu/forwarder.go | 3 +++ pkg/sfu/videolayerselector/ddforwarder.go | 21 +++++++++++++++++-- .../videolayerselector/ddforwarder_test.go | 18 ++++++++++++++++ 3 files changed, 40 insertions(+), 2 deletions(-) diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index d15fe5cdd..a413ad584 100644 --- a/pkg/sfu/forwarder.go +++ b/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 } diff --git a/pkg/sfu/videolayerselector/ddforwarder.go b/pkg/sfu/videolayerselector/ddforwarder.go index 2080f34d7..5c6fb8391 100644 --- a/pkg/sfu/videolayerselector/ddforwarder.go +++ b/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 } } diff --git a/pkg/sfu/videolayerselector/ddforwarder_test.go b/pkg/sfu/videolayerselector/ddforwarder_test.go index 701fe2503..66a155e7f 100644 --- a/pkg/sfu/videolayerselector/ddforwarder_test.go +++ b/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())