From cfedcc71d04180d6f76833940ec9f4f2e9f7bcaf Mon Sep 17 00:00:00 2001 From: CloudWebRTC Date: Fri, 19 Jun 2026 20:12:54 +0800 Subject: [PATCH] feat: acquire requested video layer directly at HIGH quality by default (#4595) * feat: acquire requested video layer directly at HIGH quality by default Two changes that together remove the visible low->high quality ramp for a new subscriber (both publisher-first and subscriber-first join orders): 1. Default a subscriber's initial video quality to HIGH on bind instead of LOW for adaptive stream, so the subscribed max layer is the top layer. Adaptive stream clients can still scale down afterwards based on viewport. 2. On initial layer acquisition the forwarder/selector latch directly onto the allocator's target (the requested top layer) instead of opportunistically latching onto the first lower key frame that arrives. A short initial-acquisition grace aims the target at the requested layer; if it does not show up in time, the target falls back to the highest layer seen so acquisition never stalls. Always on - no configuration flag. Co-Authored-By: Claude Opus 4.8 (1M context) * feat: gate start-at-desired-quality behind EnableStartAtDesiredQuality flag Put the "acquire requested video layer directly at HIGH quality" behavior behind a per-subscriber EnableStartAtDesiredQuality flag (default off, so the original low->high ramp-up is restored unless enabled). Plumbed from config.RTC.EnableStartAtDesiredQuality through ParticipantParams -> SubscribedTrack/DownTrack -> Forwarder -> simulcast selector, gating all three behavior changes: the HIGH default on bind, the forwarder's initial-acquisition grace, and the selector's direct-latch-onto-target. Co-Authored-By: Claude Opus 4.8 (1M context) * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * remove config. --------- Co-authored-by: Claude Opus 4.8 (1M context) Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- pkg/rtc/mediatracksubscriptions.go | 19 ++--- pkg/rtc/participant.go | 5 ++ pkg/rtc/subscribedtrack.go | 9 ++- pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 63 +++++++++++++++ pkg/sfu/downtrack.go | 11 +++ pkg/sfu/forwarder.go | 65 +++++++++++++++- pkg/sfu/forwarder_test.go | 42 ++++++++++ pkg/sfu/videolayerselector/base.go | 8 ++ pkg/sfu/videolayerselector/simulcast.go | 32 ++++++-- pkg/sfu/videolayerselector/simulcast_test.go | 77 +++++++++++++++++++ .../videolayerselector/videolayerselector.go | 1 + 12 files changed, 314 insertions(+), 19 deletions(-) create mode 100644 pkg/sfu/videolayerselector/simulcast_test.go diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 69575fe6b..c8720eade 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -104,15 +104,16 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * t.subscribedTracksMu.Unlock() subTrack, err := NewSubscribedTrack(SubscribedTrackParams{ - ReceiverConfig: t.params.ReceiverConfig, - SubscriberConfig: t.params.SubscriberConfig, - Subscriber: sub, - MediaTrack: t.params.MediaTrack, - AdaptiveStream: sub.GetAdaptiveStream(), - TelemetryListener: sub.GetTelemetryListener(), - WrappedReceiver: wr, - IsRelayed: t.params.IsRelayed, - OnDownTrackCreated: t.onDownTrackCreated, + ReceiverConfig: t.params.ReceiverConfig, + SubscriberConfig: t.params.SubscriberConfig, + Subscriber: sub, + MediaTrack: t.params.MediaTrack, + AdaptiveStream: sub.GetAdaptiveStream(), + EnableStartAtDesiredQuality: sub.GetEnableStartAtDesiredQuality(), + TelemetryListener: sub.GetTelemetryListener(), + WrappedReceiver: wr, + IsRelayed: t.params.IsRelayed, + OnDownTrackCreated: t.onDownTrackCreated, OnDownTrackClosed: func(subscriberID livekit.ParticipantID) { t.subscribedTracksMu.Lock() delete(t.subscribedTracks, subscriberID) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index e65733d28..ebe59cc58 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -225,6 +225,7 @@ type ParticipantParams struct { EnableRTPStreamRestartDetection bool ForceBackupCodecPolicySimulcast bool DisableTransceiverReuseForE2EE bool + EnableStartAtDesiredQuality bool } type ParticipantImpl struct { @@ -505,6 +506,10 @@ func (p *ParticipantImpl) GetAdaptiveStream() bool { return p.params.AdaptiveStream } +func (p *ParticipantImpl) GetEnableStartAtDesiredQuality() bool { + return p.params.EnableStartAtDesiredQuality +} + func (p *ParticipantImpl) GetPacer() pacer.Pacer { return p.TransportManager.GetSubscriberPacer() } diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 3a8a06148..9453bdf7d 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -48,6 +48,7 @@ type SubscribedTrackParams struct { Subscriber types.LocalParticipant MediaTrack types.MediaTrack AdaptiveStream bool + EnableStartAtDesiredQuality bool TelemetryListener types.ParticipantTelemetryListener WrappedReceiver *WrappedReceiver IsRelayed bool @@ -154,6 +155,7 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error) RTCPWriter: params.Subscriber.WriteSubscriberRTCP, DisableSenderReportPassThrough: params.Subscriber.GetDisableSenderReportPassThrough(), SupportsCodecChange: params.Subscriber.SupportsCodecChange(), + EnableStartAtDesiredQuality: params.EnableStartAtDesiredQuality, Listener: s, }) if err != nil { @@ -212,7 +214,12 @@ func (t *SubscribedTrack) Bound(err error) { t.logger.Debugw("enabling subscriber track settings on bind", "settings", logger.Proto(t.settings)) } } else { - if t.params.AdaptiveStream { + if t.params.EnableStartAtDesiredQuality { + // default to HIGH quality so the subscriber acquires the top layer directly instead of + // ramping up from a lower layer. adaptive stream clients can still scale down afterwards + // based on viewport. + t.settings = &livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH} + } else if t.params.AdaptiveStream { t.settings = &livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_LOW} } else { t.settings = &livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 194f529dd..ec10fb396 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -403,6 +403,7 @@ type LocalParticipant interface { GetReporter() roomobs.ParticipantSessionReporter GetReporterResolver() roomobs.ParticipantReporterResolver GetAdaptiveStream() bool + GetEnableStartAtDesiredQuality() bool ProtocolVersion() ProtocolVersion SupportsSyncStreamID() bool SupportsTransceiverReuse(mt MediaTrack) bool diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 76c32d76b..9570a28c2 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -325,6 +325,16 @@ type FakeLocalParticipant struct { getDisableSenderReportPassThroughReturnsOnCall map[int]struct { result1 bool } + GetEnableStartAtDesiredQualityStub func() bool + getEnableStartAtDesiredQualityMutex sync.RWMutex + getEnableStartAtDesiredQualityArgsForCall []struct { + } + getEnableStartAtDesiredQualityReturns struct { + result1 bool + } + getEnableStartAtDesiredQualityReturnsOnCall map[int]struct { + result1 bool + } GetEnabledPublishCodecsStub func() []*livekit.Codec getEnabledPublishCodecsMutex sync.RWMutex getEnabledPublishCodecsArgsForCall []struct { @@ -3079,6 +3089,59 @@ func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughReturnsOnCall }{result1} } +func (fake *FakeLocalParticipant) GetEnableStartAtDesiredQuality() bool { + fake.getEnableStartAtDesiredQualityMutex.Lock() + ret, specificReturn := fake.getEnableStartAtDesiredQualityReturnsOnCall[len(fake.getEnableStartAtDesiredQualityArgsForCall)] + fake.getEnableStartAtDesiredQualityArgsForCall = append(fake.getEnableStartAtDesiredQualityArgsForCall, struct { + }{}) + stub := fake.GetEnableStartAtDesiredQualityStub + fakeReturns := fake.getEnableStartAtDesiredQualityReturns + fake.recordInvocation("GetEnableStartAtDesiredQuality", []interface{}{}) + fake.getEnableStartAtDesiredQualityMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetEnableStartAtDesiredQualityCallCount() int { + fake.getEnableStartAtDesiredQualityMutex.RLock() + defer fake.getEnableStartAtDesiredQualityMutex.RUnlock() + return len(fake.getEnableStartAtDesiredQualityArgsForCall) +} + +func (fake *FakeLocalParticipant) GetEnableStartAtDesiredQualityCalls(stub func() bool) { + fake.getEnableStartAtDesiredQualityMutex.Lock() + defer fake.getEnableStartAtDesiredQualityMutex.Unlock() + fake.GetEnableStartAtDesiredQualityStub = stub +} + +func (fake *FakeLocalParticipant) GetEnableStartAtDesiredQualityReturns(result1 bool) { + fake.getEnableStartAtDesiredQualityMutex.Lock() + defer fake.getEnableStartAtDesiredQualityMutex.Unlock() + fake.GetEnableStartAtDesiredQualityStub = nil + fake.getEnableStartAtDesiredQualityReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) GetEnableStartAtDesiredQualityReturnsOnCall(i int, result1 bool) { + fake.getEnableStartAtDesiredQualityMutex.Lock() + defer fake.getEnableStartAtDesiredQualityMutex.Unlock() + fake.GetEnableStartAtDesiredQualityStub = nil + if fake.getEnableStartAtDesiredQualityReturnsOnCall == nil { + fake.getEnableStartAtDesiredQualityReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.getEnableStartAtDesiredQualityReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) GetEnabledPublishCodecs() []*livekit.Codec { fake.getEnabledPublishCodecsMutex.Lock() ret, specificReturn := fake.getEnabledPublishCodecsReturnsOnCall[len(fake.getEnabledPublishCodecsArgsForCall)] diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 38cf31569..b1a516799 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -312,6 +312,7 @@ type DownTrackParams struct { DisableSenderReportPassThrough bool SupportsCodecChange bool StripPacketTrailer bool + EnableStartAtDesiredQuality bool Listener DownTrackListener } @@ -463,6 +464,7 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) { d.params.Logger, false, // skipReferenceTS false, // disableOpportunisticAllocation + d.params.EnableStartAtDesiredQuality, d.rtpStats, ) @@ -1026,6 +1028,15 @@ func (d *DownTrack) keyFrameRequester() { d.Receiver().SendPLI(layer, false) d.rtpStats.UpdateLayerLockPliAndTime(1) } + + // if the initial-acquisition grace expired without latching the requested layer, force a + // re-allocation so the target falls back to the highest layer actually seen (rather than + // stalling while waiting for a requested layer that never showed up) + if d.forwarder.MaybeExpireAcquireGrace() { + if sal := d.getStreamAllocatorListener(); sal != nil { + sal.OnAvailableLayersChanged(d) + } + } } } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 4fac35962..3bfae7d4b 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -52,6 +52,15 @@ const ( ResumeBehindHighThresholdSeconds = float64(2.0) // 2 seconds LayerSwitchBehindThresholdSeconds = float64(0.05) // 50ms SwitchAheadThresholdSeconds = float64(0.025) // 25ms + + // While a subscriber is acquiring its first layer and the requested (max) layer has not been + // seen on the wire yet, aim straight for the requested layer for this long instead of latching + // onto a lower layer that is detected first. Avoids a visible low -> high quality ramp, + // notably when the subscriber joined before the publisher started (so layers are detected, and + // `maxSeen` climbs, gradually). If the requested layer does not show up within this window the + // grace expires and forwarding falls back to the highest layer actually seen. + // See Forwarder.opportunisticAlloc / withinAcquireGraceLocked / MaybeExpireAcquireGrace. + initialLayerAcquisitionGrace = time.Second ) var ( @@ -222,6 +231,7 @@ type Forwarder struct { logger logger.Logger skipReferenceTS bool disableOpportunisticAllocation bool + enableStartAtDesiredQuality bool rtpStats *rtpstats.RTPStatsSender muted bool @@ -230,6 +240,7 @@ type Forwarder struct { started bool preStartTime time.Time + acquireDeadline int64 // mono nanos; initial-acquisition grace deadline, 0 = inactive extFirstTS uint64 lastSSRC uint32 lastReferencePayloadType int8 @@ -256,6 +267,7 @@ func NewForwarder( logger logger.Logger, skipReferenceTS bool, disableOpportunisticAllocation bool, + enableStartAtDesiredQuality bool, rtpStats *rtpstats.RTPStatsSender, ) *Forwarder { f := &Forwarder{ @@ -264,6 +276,7 @@ func NewForwarder( logger: logger, skipReferenceTS: skipReferenceTS, disableOpportunisticAllocation: disableOpportunisticAllocation, + enableStartAtDesiredQuality: enableStartAtDesiredQuality, rtpStats: rtpStats, referenceLayerSpatial: buffer.InvalidLayerSpatial, lastAllocation: VideoAllocationDefault, @@ -276,6 +289,7 @@ func NewForwarder( if f.kind == webrtc.RTPCodecTypeVideo { f.vls.SetMaxTemporal(buffer.DefaultMaxLayerTemporal) } + f.vls.SetEnableStartAtDesiredQuality(enableStartAtDesiredQuality) return f } @@ -289,10 +303,36 @@ func (f *Forwarder) SetMaxPublishedLayer(maxPublishedLayer int32) bool { } f.vls.SetMaxSeenSpatial(maxPublishedLayer) + if f.enableStartAtDesiredQuality && !f.vls.GetCurrent().IsValid() { + // A (higher) layer just became available while nothing is being forwarded yet. + // (Re)start the initial-acquisition grace so the target aims for the requested layer + // instead of ramping up gradually as more layers are detected. See opportunisticAlloc. + f.acquireDeadline = mono.UnixNano() + initialLayerAcquisitionGrace.Nanoseconds() + } f.logger.Debugw("setting max published layer", "layer", maxPublishedLayer) return true } +// withinAcquireGraceLocked reports whether the initial-acquisition grace window is still open. +func (f *Forwarder) withinAcquireGraceLocked() bool { + return f.acquireDeadline != 0 && mono.UnixNano() < f.acquireDeadline +} + +// MaybeExpireAcquireGrace returns true once when the initial-acquisition grace has expired while +// the forwarder is still not streaming any layer. The caller should trigger a re-allocation so the +// target falls back from the requested layer to the highest layer actually seen, avoiding a stall +// if the requested layer never shows up. The deadline is cleared so it fires at most once. +func (f *Forwarder) MaybeExpireAcquireGrace() bool { + f.lock.Lock() + defer f.lock.Unlock() + + if f.acquireDeadline == 0 || mono.UnixNano() < f.acquireDeadline { + return false + } + f.acquireDeadline = 0 + return !f.vls.GetCurrent().IsValid() +} + func (f *Forwarder) SetMaxTemporalLayerSeen(maxTemporalLayerSeen int32) bool { f.lock.Lock() defer f.lock.Unlock() @@ -832,6 +872,17 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow return maxTemporal } + // inAcquireGrace reports that we are acquiring the first layer, the requested layer has not + // been seen on the wire yet, and the grace window is still open. While true, aim straight for + // the requested layer instead of the highest seen so far, so acquisition does not ramp up + // gradually as layers are detected (`maxSeen` climbs). See initialLayerAcquisitionGrace. + inAcquireGrace := func(maxSpatial int32) bool { + return !currentLayer.IsValid() && maxSeenLayer.Spatial < maxSpatial && f.withinAcquireGraceLocked() + } + + // set when opportunisticAlloc aimed the target at the requested layer due to the acquisition + // grace (as opposed to overshoot), so the key frame request can be pointed at it too + acquireGraceApplied := false opportunisticAlloc := func() { // opportunistically latch on to anything maxSpatial := maxLayer.Spatial @@ -839,8 +890,14 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow maxSpatial = maxSeenLayer.Spatial } + targetSpatial := min(maxSeenLayer.Spatial, maxSpatial) + if inAcquireGrace(maxSpatial) { + targetSpatial = maxSpatial + acquireGraceApplied = true + } + alloc.TargetLayer = buffer.VideoLayer{ - Spatial: min(maxSeenLayer.Spatial, maxSpatial), + Spatial: targetSpatial, Temporal: getMaxTemporal(), } } @@ -935,7 +992,11 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow } else { // opportunistically latch on to anything opportunisticAlloc() - if requestLayerSpatial == buffer.InvalidLayerSpatial { + if acquireGraceApplied { + // in the acquisition grace, request a key frame for the requested layer we + // are waiting for (above what has been seen so far) + alloc.RequestLayerSpatial = alloc.TargetLayer.Spatial + } else if requestLayerSpatial == buffer.InvalidLayerSpatial { alloc.RequestLayerSpatial = maxLayerSpatialLimit } else { alloc.RequestLayerSpatial = requestLayerSpatial diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index cca1e69be..513fbe40e 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -39,6 +39,7 @@ func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Fo logger.GetLogger(), true, // skipReferenceTS true, // disableOpportunisticAllocation + true, // enableStartAtDesiredQuality nil, ) f.DetermineCodec(codec, nil, livekit.VideoLayer_MODE_UNUSED) @@ -2145,3 +2146,44 @@ func TestForwarderGetPaddingVP8(t *testing.T) { require.NoError(t, err) require.Equal(t, marshalledVP8, buf) } + +func TestForwarderInitialAcquisitionGrace(t *testing.T) { + f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + + // subscriber requested the top spatial layer + f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial) + f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal) + f.SetMaxTemporalLayerSeen(buffer.DefaultMaxLayerTemporal) + + bitrates := Bitrates{ + {2, 3, 0, 0}, + {4, 0, 0, 5}, + {0, 7, 0, 0}, + } + + // subscriber-first: only layer 1 has been seen so far and nothing is being forwarded yet + // (current invalid). this arms the initial-acquisition grace. + require.True(t, f.SetMaxPublishedLayer(1)) + f.lock.RLock() + require.True(t, f.withinAcquireGraceLocked()) + f.lock.RUnlock() + + // during the grace, the target aims straight at the requested layer (2) and requests a key + // frame for it, even though only layer 1 has been seen - so acquisition does not ramp up + // gradually as higher layers are detected + alloc := f.AllocateOptimal([]int32{0, 1}, bitrates, false, false) + require.Equal(t, int32(2), alloc.TargetLayer.Spatial) + require.Equal(t, int32(2), alloc.RequestLayerSpatial) + + // force the grace to expire while still not streaming: must signal that a re-allocation is + // needed so the target can fall back + f.acquireDeadline = 1 // a deadline far in the past + require.True(t, f.MaybeExpireAcquireGrace()) + require.False(t, f.MaybeExpireAcquireGrace()) // only fires once + + // after the grace, the target falls back to the highest layer actually seen (1) instead of + // stalling while waiting for a requested layer that never showed up + alloc = f.AllocateOptimal([]int32{0, 1}, bitrates, false, false) + require.Equal(t, int32(1), alloc.TargetLayer.Spatial) + require.Equal(t, int32(1), alloc.RequestLayerSpatial) +} diff --git a/pkg/sfu/videolayerselector/base.go b/pkg/sfu/videolayerselector/base.go index 181e5d3fb..9fc74c16a 100644 --- a/pkg/sfu/videolayerselector/base.go +++ b/pkg/sfu/videolayerselector/base.go @@ -35,6 +35,10 @@ type Base struct { currentLayer buffer.VideoLayer previousLayer buffer.VideoLayer + + // when set, on initial acquisition latch directly onto the target (requested) layer instead of + // opportunistically latching onto the first lower-layer key frame that arrives (see Simulcast.Select) + enableStartAtDesiredQuality bool } func NewBase(logger logger.Logger) *Base { @@ -66,6 +70,10 @@ func (b *Base) SetTemporalLayerSelector(tls temporallayerselector.TemporalLayerS b.tls = tls } +func (b *Base) SetEnableStartAtDesiredQuality(enable bool) { + b.enableStartAtDesiredQuality = enable +} + func (b *Base) SetMax(maxLayer buffer.VideoLayer) { b.maxLayer = maxLayer } diff --git a/pkg/sfu/videolayerselector/simulcast.go b/pkg/sfu/videolayerselector/simulcast.go index 664b2a706..9f0770b7a 100644 --- a/pkg/sfu/videolayerselector/simulcast.go +++ b/pkg/sfu/videolayerselector/simulcast.go @@ -94,14 +94,32 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL found := false reason := "" if extPkt.IsKeyFrame { - if layer > s.currentLayer.Spatial && layer <= s.targetLayer.Spatial { - reason = "upgrading layer" - found = true - } + if s.enableStartAtDesiredQuality && !isActive { + // Initial acquisition: latch directly onto the target layer instead of + // opportunistically latching onto the first key frame of any lower layer that + // happens to arrive first. This avoids a visible low-quality -> high-quality ramp + // (e.g. briefly decoding layer 0 before settling on a requested layer 2) for a + // subscriber that requested the higher layer. + // + // The target is chosen by the allocator: during the initial-acquisition grace it + // points at the requested layer (so we wait for it); if that layer never shows + // up the grace expires and the allocator drops the target to the highest layer + // actually seen, so we always end up latching onto a layer that is flowing. + if layer == s.targetLayer.Spatial { + reason = "acquiring target layer" + found = true + } + } else { + // default: opportunistically latch on to / step towards the target layer + if layer > s.currentLayer.Spatial && layer <= s.targetLayer.Spatial { + reason = "upgrading layer" + found = true + } - if layer < s.currentLayer.Spatial && layer >= s.targetLayer.Spatial { - reason = "downgrading layer" - found = true + if layer < s.currentLayer.Spatial && layer >= s.targetLayer.Spatial { + reason = "downgrading layer" + found = true + } } if found { diff --git a/pkg/sfu/videolayerselector/simulcast_test.go b/pkg/sfu/videolayerselector/simulcast_test.go new file mode 100644 index 000000000..878db5eec --- /dev/null +++ b/pkg/sfu/videolayerselector/simulcast_test.go @@ -0,0 +1,77 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package videolayerselector + +import ( + "testing" + + "github.com/pion/rtp" + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/protocol/logger" +) + +func keyFrameOnLayer(spatial, temporal int32) *buffer.ExtPacket { + return &buffer.ExtPacket{ + VideoLayer: buffer.VideoLayer{Spatial: spatial, Temporal: temporal}, + Packet: &rtp.Packet{}, + IsKeyFrame: true, + } +} + +// On initial acquisition the selector must latch directly onto the target layer and ignore +// lower-layer key frames that arrive first, so a subscriber requesting the top layer does not +// briefly decode a lower layer (a visible quality ramp). +func TestSimulcastSelectAcquiresTargetLayerDirectly(t *testing.T) { + s := NewSimulcast(logger.GetLogger()) + s.SetEnableStartAtDesiredQuality(true) + s.SetMax(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetMaxSeen(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetTarget(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetRequestSpatial(2) + s.SetCurrent(buffer.InvalidLayer) + + // lower-layer key frames arriving first must NOT be latched + require.False(t, s.Select(keyFrameOnLayer(0, 2), 0).IsSelected) + require.False(t, s.GetCurrent().IsValid()) + require.False(t, s.Select(keyFrameOnLayer(1, 2), 1).IsSelected) + require.False(t, s.GetCurrent().IsValid()) + + // the target layer key frame latches directly + require.True(t, s.Select(keyFrameOnLayer(2, 2), 2).IsSelected) + require.Equal(t, int32(2), s.GetCurrent().Spatial) +} + +// Acquisition follows whatever target the allocator set: when the target is lowered (e.g. the +// acquisition grace expired and the allocator fell back to the highest layer seen), the selector +// latches that lower layer directly. This is the fallback path that prevents a stall when the +// originally requested layer never shows up. +func TestSimulcastSelectAcquiresLoweredTarget(t *testing.T) { + s := NewSimulcast(logger.GetLogger()) + s.SetEnableStartAtDesiredQuality(true) + s.SetMax(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetMaxSeen(buffer.VideoLayer{Spatial: 1, Temporal: 2}) + // allocator dropped the target to the highest layer actually seen + s.SetTarget(buffer.VideoLayer{Spatial: 1, Temporal: 2}) + s.SetRequestSpatial(1) + s.SetCurrent(buffer.InvalidLayer) + + require.False(t, s.Select(keyFrameOnLayer(0, 2), 0).IsSelected) + require.False(t, s.GetCurrent().IsValid()) + + require.True(t, s.Select(keyFrameOnLayer(1, 2), 1).IsSelected) + require.Equal(t, int32(1), s.GetCurrent().Spatial) +} diff --git a/pkg/sfu/videolayerselector/videolayerselector.go b/pkg/sfu/videolayerselector/videolayerselector.go index 883e46aea..72a6e2c81 100644 --- a/pkg/sfu/videolayerselector/videolayerselector.go +++ b/pkg/sfu/videolayerselector/videolayerselector.go @@ -37,6 +37,7 @@ type VideoLayerSelector interface { IsOvershootOkay() bool SetTemporalLayerSelector(tls temporallayerselector.TemporalLayerSelector) + SetEnableStartAtDesiredQuality(enable bool) SetMax(maxLayer buffer.VideoLayer) SetMaxSpatial(layer int32)