diff --git a/pkg/clientconfiguration/conf_test.go b/pkg/clientconfiguration/conf_test.go index e11738750..00b14ffe1 100644 --- a/pkg/clientconfiguration/conf_test.go +++ b/pkg/clientconfiguration/conf_test.go @@ -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"} diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 00fe9d896..d15fe5cdd 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -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", diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 3d32c1a0a..81da3a06a 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -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) +} diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index 724f04e26..091189d0c 100644 --- a/pkg/sfu/sequencer.go +++ b/pkg/sfu/sequencer.go @@ -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 }