From 103d7bec847a21581e3d4d8ac5518330bf22196f Mon Sep 17 00:00:00 2001 From: Miguel Delgado <114915581+miguel-teamsquad@users.noreply.github.com> Date: Fri, 16 Oct 2026 14:54:18 +0200 Subject: [PATCH] sfu: apply dummy start offset modulo 2^32 on source switch (#5018) processSourceSwitch added dummyStartTSOffset to extRefTS in 64 bits. When the offset is negative and the reference layer's 32-bit RTP timestamp has wrapped but is still below |offset|, the addition borrows from the upper half and extRefTS ends up 2^32 below the right value. The wrap correction that follows only compares the lower 32 bits, so it cannot undo it. The switch point then looks ~47,722 s (2^32 / 90 kHz) behind and is rejected with errSwitchPointTooFarBehind on every keyframe, until the reference layer's timestamp climbs past |offset|, which can take hours. Apply the offset modulo 2^32 before restoring the cycles from extLastTS. The lower 32 bits, and so the timestamp sent to subscribers, are unchanged. --- pkg/sfu/forwarder.go | 11 ++- pkg/sfu/forwarder_test.go | 154 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 163 insertions(+), 2 deletions(-) diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 8e2786c08..00fe9d896 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -2003,9 +2003,16 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) ( } // adjust extRefTS to current packet's timestamp mapped to that of reference layer's - extRefTS = (extRefTS & 0xFFFF_FFFF_0000_0000) + uint64(refTS) + f.dummyStartTSOffset + // + // The dummy start offset is applied modulo 2^32, before the cycles are restored from + // extLastTS. Adding it in 64 bits lets a borrow (negative offset) or a carry (positive offset) + // reach the upper half of extRefTS, and the wrap correction below only compares the lower 32 + // bits, so it cannot undo it: extRefTS would end up 2^32 below (or above) the right value. + // With a negative offset that rejects every switch point as too far behind until the + // reference layer's 32-bit timestamp climbs past |dummyStartTSOffset|, which can take hours. + refTS = uint32(uint64(refTS) + f.dummyStartTSOffset) + extRefTS = (extRefTS & 0xFFFF_FFFF_0000_0000) + uint64(refTS) lastTS := uint32(extLastTS) - refTS = uint32(extRefTS) if (refTS-lastTS) < 1<<31 && refTS < lastTS { extRefTS += (1 << 32) } diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 567a79096..3d32c1a0a 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -16,12 +16,15 @@ package sfu import ( "encoding/hex" + "math/rand" "testing" + "time" "github.com/pion/rtp/codecs" "github.com/pion/webrtc/v4" "github.com/stretchr/testify/require" + "github.com/livekit/mediatransportutil" "github.com/livekit/mediatransportutil/pkg/codec" "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/livekit" @@ -31,6 +34,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/codecmunger" "github.com/livekit/livekit-server/pkg/sfu/pacer" dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor" + "github.com/livekit/livekit-server/pkg/sfu/rtpstats" "github.com/livekit/livekit-server/pkg/sfu/testutils" "github.com/livekit/livekit-server/pkg/sfu/videolayerselector" ) @@ -2891,3 +2895,153 @@ func TestForwarderSimulcastDependencyDescriptor(t *testing.T) { }) } } + +// --------------------------------------------------------------------------- +// Switching back to a layer after the reference layer's RTP timestamp wrapped, +// on a downtrack that started with dummy forwarding (dummyStartTSOffset != 0). + +const ( + wrapTestSSRCLow = uint32(0x0a0a0a0a) + wrapTestSSRCHigh = uint32(0x0c0c0c0c) + wrapTestFrame = uint32(3000) // 30 fps at 90 kHz + wrapTestHighTS = uint32(0x10000000) +) + +func newWrapTestForwarder(t *testing.T) *Forwarder { + t.Helper() + f := NewForwarder( + webrtc.RTPCodecTypeVideo, + logger.GetLogger(), + false, // skipReferenceTS + false, // disableOpportunisticAllocation + false, // enableStartAtDesiredQuality + rtpstats.NewRTPStatsSender(rtpstats.RTPStatsParams{}, 1024), + ) + f.DetermineCodec(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000}, nil, livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM) + f.SetMaxSpatialLayer(2) + f.SetMaxTemporalLayer(0) + f.SetMaxPublishedLayer(2) + return f +} + +func wrapTestBitrates(layer int32) Bitrates { + var brs Bitrates + brs[layer][0] = 1_000_000 + return brs +} + +func wrapTestKeyFrame(t *testing.T, ssrc uint32, sn uint16, ts uint32, tsCycles int, layer int32) *buffer.ExtPacket { + t.Helper() + extPkt, err := testutils.GetTestExtPacket(&testutils.TestExtPacketParams{ + SequenceNumber: sn, + Timestamp: ts, + TSCycles: tsCycles, + SSRC: ssrc, + PayloadSize: 20, + Marker: true, + IsKeyFrame: true, + VideoLayer: buffer.VideoLayer{Spatial: layer, Temporal: 0}, + }) + require.NoError(t, err) + return extPkt +} + +func wrapTestSenderReport(at time.Time, rtpTS uint32) *livekit.RTCPSenderReportState { + return &livekit.RTCPSenderReportState{ + RtpTimestamp: rtpTS, + RtpTimestampExt: uint64(rtpTS), + NtpTimestamp: uint64(mediatransportutil.ToNtpTime(at)), + At: at.UnixNano(), + AtAdjusted: at.UnixNano(), + } +} + +// runToHighLayer starts the forwarder with dummy forwarding, locks it onto layer 0 (whose first +// timestamp is lowStart), switches it to layer 2 when layer 0 stops, forwards layer 2 for +// `elapsed`, and then has the allocator target layer 0 again. It returns layer 0's timestamp and +// timestamp cycles at that point, for the keyframe that switches back. +func runToHighLayer(t *testing.T, f *Forwarder, lowStart uint32, elapsed time.Duration) (uint32, int) { + t.Helper() + + // dummy start, then lock onto layer 0: layer 0 becomes the reference layer + f.maybeStart() + f.AllocateOptimal([]int32{0}, wrapTestBitrates(0), true, false) + tp, err := f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCLow, 1000, lowStart, 0, 0), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.Equal(t, int32(0), f.referenceLayerSpatial) + require.Equal(t, int32(0), f.CurrentLayer().Spatial) + + now := time.Now() + f.SetRefSenderReport(0, wrapTestSenderReport(now, lowStart+wrapTestFrame)) + f.SetRefSenderReport(2, wrapTestSenderReport(now, wrapTestHighTS+wrapTestFrame)) + + // layer 0 stops, layer 2 is available: switch to layer 2 + alloc := f.AllocateOptimal([]int32{2}, wrapTestBitrates(2), true, false) + require.Equal(t, int32(2), alloc.TargetLayer.Spatial) + tp, err = f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCHigh, 5000, wrapTestHighTS+2*wrapTestFrame, 0, 2), 2) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.Equal(t, int32(2), f.CurrentLayer().Spatial) + + // forward layer 2 for `elapsed` + ticks := uint64(elapsed.Seconds() * 90000) + highTS := uint64(wrapTestHighTS) + ticks + tp, err = f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCHigh, 5001, uint32(highTS), int(highTS>>32), 2), 2) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + + // layer 2 stops, layer 0 is back: the allocator targets layer 0 + alloc = f.AllocateOptimal([]int32{0}, wrapTestBitrates(0), true, false) + require.Equal(t, int32(0), alloc.TargetLayer.Spatial) + + lowTS := uint64(lowStart) + ticks + uint64(wrapTestFrame) + return uint32(lowTS), int(lowTS >> 32) +} + +func TestForwarderSwitchBackAfterReferenceTimestampWrap(t *testing.T) { + // maybeStart draws the dummy first timestamp in [2^31, 2^31 + 2^30), so a reference layer + // starting above 2^31 + 2^30 always yields a negative dummyStartTSOffset + const lowStart = uint32(0xf0000000) + + t.Run("after the wrap", func(t *testing.T) { + f := newWrapTestForwarder(t) + lowTS, cycles := runToHighLayer(t, f, lowStart, time.Hour) + + // preconditions: negative offset, layer 0 wrapped and its 32-bit timestamp is below |offset| + offset := int64(f.dummyStartTSOffset) + require.Less(t, offset, int64(0)) + require.Equal(t, 1, cycles) + require.Less(t, int64(lowTS), -offset) + + tp, err := f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCLow, 1001, lowTS, cycles, 0), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop, "switch point rejected") + require.Equal(t, int32(0), f.CurrentLayer().Spatial) + }) + + t.Run("before the wrap", func(t *testing.T) { + f := newWrapTestForwarder(t) + lowTS, cycles := runToHighLayer(t, f, lowStart, 30*time.Minute) + require.Equal(t, 0, cycles) + + tp, err := f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCLow, 1001, lowTS, cycles, 0), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop, "switch point rejected") + require.Equal(t, int32(0), f.CurrentLayer().Spatial) + }) + + t.Run("random start and duration", func(t *testing.T) { + rng := rand.New(rand.NewSource(1)) + for i := 0; i < 400; i++ { + f := newWrapTestForwarder(t) + start := rng.Uint32() + elapsed := time.Duration(60+rng.Int63n(26*3600)) * time.Second + lowTS, cycles := runToHighLayer(t, f, start, elapsed) + + tp, err := f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCLow, 1001, lowTS, cycles, 0), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop, "switch point rejected, start %d, elapsed %s", start, elapsed) + } + }) +}