dandri/livekit
1
0 Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-23 04:22:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

sfu: reset dummy start offset on forwarder restart (#5019)

Browse Source 
* sfu: reset dummy start offset on forwarder restart

Restart() kept dummyStartTSOffset. The next source switch recomputed it
from a refTS that already included the old offset, so every switch after
that was off by the old value and the subscriber's timestamp jumped.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* sfu: log extTimestamp in sequencer packet meta, add Chrome case to AV1 config test

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* sfu: do not apply dummy start offset without a reference layer

With skipReferenceTS, refTS is the last outgoing timestamp, which is
already on the downtrack's timeline. Computing and adding an offset there
re-added the gap between the last packet and a restart on every later
layer switch.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Raja SubramanianandClaude Fable 5.1 authored and GitHub committed 2026-10-16 19:04:20 +05:30
1 parent 103d7bec84
commit cde3991916
4 files changed
+114 -17
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/clientconfiguration/conf_test.go
+2
View File
@@ -67,6 +67,8 @@ func TestStaticConfigurationAV1(t *testing.T) {
require.True(t, av1Disabled(&livekit.ClientInfo{Browser: "firefox", BrowserVersion: "158.0.2", Os: "windows"}))
// an unknown version is treated as affected
require.True(t, av1Disabled(&livekit.ClientInfo{Browser: "firefox", Os: "linux"}))
// should not be disabled for Chrome
require.False(t, av1Disabled(&livekit.ClientInfo{Browser: "chrome", Os: "linux"}))
// Firefox for Android keeps the AV1 rule alongside its H.264 publish rule
android := &livekit.ClientInfo{Browser: "firefox mobile", BrowserVersion: "156.0", Os: "android"}
pkg/sfu/forwarder.go
+22 -12
View File
@@ -1807,6 +1807,10 @@ func (f *Forwarder) Restart() {
f.setTargetLayer(buffer.InvalidLayer, buffer.InvalidLayerSpatial)
f.referenceLayerSpatial = buffer.InvalidLayerSpatial
f.lastReferencePayloadType = -1
// The next source switch re-derives the offset from the new reference layer's raw timestamp,
// so that it maps onto the downtrack's existing timeline (extExpectedTS). Keeping the old
// offset would fold it into the new one and shift every later switch by the old value.
f.dummyStartTSOffset = 0
for layer := range len(f.refInfos) {
f.refInfos[layer] = refInfo{}
@@ -2000,17 +2004,19 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) (
)
return starting, err
}
}
// adjust extRefTS to current packet's timestamp mapped to that of reference layer's
//
// 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)
// adjust refTS to current packet's timestamp mapped to that of reference layer's.
// Without a reference layer (skipReferenceTS), refTS is the last outgoing timestamp,
// which is already on the downtrack's timeline and needs no offset.
//
// 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)
if (refTS-lastTS) < 1<<31 && refTS < lastTS {
@@ -2025,7 +2031,9 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) (
if err == nil {
extExpectedTS = tsExt
if f.lastReferencePayloadType == -1 {
f.dummyStartTSOffset = extExpectedTS - uint64(refTS)
if !f.skipReferenceTS {
f.dummyStartTSOffset = extExpectedTS - uint64(refTS)
}
extRefTS = extExpectedTS
}
} else {
@@ -2034,7 +2042,9 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) (
rtpDiff := uint64(timeSinceFirst.Nanoseconds() * int64(f.clockRate) / 1e9)
extExpectedTS = f.extFirstTS + rtpDiff
if f.dummyStartTSOffset == 0 {
f.dummyStartTSOffset = extExpectedTS - uint64(refTS)
if !f.skipReferenceTS {
f.dummyStartTSOffset = extExpectedTS - uint64(refTS)
}
extRefTS = extExpectedTS
f.logger.Infow(
"calculating dummyStartTSOffset",
pkg/sfu/forwarder_test.go
+89 -5
View File
@@ -29,6 +29,7 @@ import (
"github.com/livekit/protocol/codecs/mime"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/utils/mono"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/livekit-server/pkg/sfu/codecmunger"
@@ -2907,12 +2908,12 @@ const (
wrapTestHighTS = uint32(0x10000000)
)
func newWrapTestForwarder(t *testing.T) *Forwarder {
func newWrapTestForwarder(t *testing.T, skipReferenceTS bool) *Forwarder {
t.Helper()
f := NewForwarder(
webrtc.RTPCodecTypeVideo,
logger.GetLogger(),
false, // skipReferenceTS
skipReferenceTS,
false, // disableOpportunisticAllocation
false, // enableStartAtDesiredQuality
rtpstats.NewRTPStatsSender(rtpstats.RTPStatsParams{}, 1024),
@@ -3005,7 +3006,7 @@ func TestForwarderSwitchBackAfterReferenceTimestampWrap(t *testing.T) {
const lowStart = uint32(0xf0000000)
t.Run("after the wrap", func(t *testing.T) {
f := newWrapTestForwarder(t)
f := newWrapTestForwarder(t, false)
lowTS, cycles := runToHighLayer(t, f, lowStart, time.Hour)
// preconditions: negative offset, layer 0 wrapped and its 32-bit timestamp is below |offset|
@@ -3021,7 +3022,7 @@ func TestForwarderSwitchBackAfterReferenceTimestampWrap(t *testing.T) {
})
t.Run("before the wrap", func(t *testing.T) {
f := newWrapTestForwarder(t)
f := newWrapTestForwarder(t, false)
lowTS, cycles := runToHighLayer(t, f, lowStart, 30*time.Minute)
require.Equal(t, 0, cycles)
@@ -3034,7 +3035,7 @@ func TestForwarderSwitchBackAfterReferenceTimestampWrap(t *testing.T) {
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)
f := newWrapTestForwarder(t, false)
start := rng.Uint32()
elapsed := time.Duration(60+rng.Int63n(26*3600)) * time.Second
lowTS, cycles := runToHighLayer(t, f, start, elapsed)
@@ -3045,3 +3046,86 @@ func TestForwarderSwitchBackAfterReferenceTimestampWrap(t *testing.T) {
}
})
}
// Receiver restart on a downtrack that started with dummy forwarding: the new reference layer's
// timestamps must keep mapping onto the downtrack's existing timeline on every switch after
// the restart, not only on the first one.
func TestForwarderRestartKeepsTimestampContinuity(t *testing.T) {
f := newWrapTestForwarder(t, false)
f.rtpStats.SetClockRate(90000)
// dummy start, lock onto layer 0 with a reference timestamp that yields a negative offset
f.maybeStart()
f.AllocateOptimal([]int32{0}, wrapTestBitrates(0), true, false)
tp, err := f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCLow, 1000, 0xf0000000, 0, 0), 0)
require.NoError(t, err)
require.False(t, tp.shouldDrop)
require.Less(t, int64(f.dummyStartTSOffset), int64(0))
// the downtrack has been sending: rtpStats is initialized on the first forwarded packet
f.rtpStats.Update(mono.UnixNano(), tp.rtp.extSequenceNumber, tp.rtp.extTimestamp, true, 12, 20, 0, false)
// upstream receiver restart: new stream, new timestamp base for every layer
f.Restart()
f.DetermineCodec(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000}, nil, livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM)
const newLowStart = uint32(0x20000000)
now := time.Now()
f.SetRefSenderReport(0, wrapTestSenderReport(now, newLowStart))
f.SetRefSenderReport(2, wrapTestSenderReport(now, wrapTestHighTS))
f.AllocateOptimal([]int32{0}, wrapTestBitrates(0), true, false)
tp, err = f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCLow, 1, newLowStart, 0, 0), 0)
require.NoError(t, err)
require.False(t, tp.shouldDrop)
lastTS := tp.rtp.extTimestamp
// one frame later, switch to layer 2: the timestamp must advance by about one frame
f.AllocateOptimal([]int32{2}, wrapTestBitrates(2), true, false)
tp, err = f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCHigh, 1, wrapTestHighTS+wrapTestFrame, 0, 2), 2)
require.NoError(t, err)
require.False(t, tp.shouldDrop)
advance := int64(tp.rtp.extTimestamp - lastTS)
require.Greater(t, advance, int64(0))
require.Less(t, advance, int64(90000), "timestamp jumped %d ticks (%.1f s) across the switch after restart", advance, float64(advance)/90000)
}
// Stream without a usable reference layer (incomplete sender reports, skipReferenceTS): refTS is
// the last outgoing timestamp, already on the downtrack's timeline. No dummy start offset may be
// computed or applied, otherwise the gap between the last packet and the restart is added again
// on every later switch.
func TestForwarderSkipReferenceTSRestartNoOffset(t *testing.T) {
f := newWrapTestForwarder(t, true)
f.rtpStats.SetClockRate(90000)
f.maybeStart()
f.AllocateOptimal([]int32{0}, wrapTestBitrates(0), true, false)
tp, err := f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCLow, 1000, 0xf0000000, 0, 0), 0)
require.NoError(t, err)
require.False(t, tp.shouldDrop)
require.Zero(t, f.dummyStartTSOffset)
firstTS := tp.rtp.extTimestamp
// the last packet went out 5 s ago
const gap = 5 * time.Second
f.rtpStats.Update(mono.UnixNano()-gap.Nanoseconds(), tp.rtp.extSequenceNumber, firstTS, true, 12, 20, 0, false)
// upstream restart, new SSRC: the 5 s gap shows up once, at the restart
f.Restart()
f.DetermineCodec(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000}, nil, livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM_INCOMPLETE_RTCP_SR)
f.AllocateOptimal([]int32{0}, wrapTestBitrates(0), true, false)
tp, err = f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCHigh, 1, 0x20000000, 0, 0), 0)
require.NoError(t, err)
require.False(t, tp.shouldDrop)
require.Zero(t, f.dummyStartTSOffset)
require.InDelta(t, gap.Seconds()*90000, float64(int64(tp.rtp.extTimestamp-firstTS)), 90000)
lastTS := tp.rtp.extTimestamp
// one frame later, switch to another layer: the gap must not be added again
f.AllocateOptimal([]int32{2}, wrapTestBitrates(2), true, false)
tp, err = f.GetTranslationParams(wrapTestKeyFrame(t, wrapTestSSRCLow, 1, wrapTestHighTS+wrapTestFrame, 0, 2), 2)
require.NoError(t, err)
require.False(t, tp.shouldDrop)
advance := int64(tp.rtp.extTimestamp - lastTS)
require.GreaterOrEqual(t, advance, int64(0))
require.Less(t, advance, int64(90000), "timestamp jumped %d ticks (%.1f s) on the switch after restart", advance, float64(advance)/90000)
}
pkg/sfu/sequencer.go
+1
View File
@@ -106,6 +106,7 @@ type extPacketMeta struct {
func (epm extPacketMeta) MarshalLogObject(e zapcore.ObjectEncoder) error {
e.AddObject("packetMeta", epm.packetMeta)
e.AddUint64("extSequenceNumber", epm.extSequenceNumber)
e.AddUint64("extTimestamp", epm.extTimestamp)
return nil
}