diff --git a/config-sample.yaml b/config-sample.yaml index ddb938353..68ea80813 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -122,6 +122,12 @@ rtc: # # forwarding. # flexfec: # upstream_enabled: false + # # Allow repairs for subscribers that negotiate FlexFEC. Clients select fec + # # per track in UpdateTrackSettings: FEC_NONE (default), FEC_LOW (15%), + # # FEC_MEDIUM (25%), or FEC_HIGH (35%) repair packet overhead, matching publish + # # track presets. Media is sent immediately; incomplete groups + # # spanning more than 200ms and packets without room for repair headers are skipped. + # downstream_enabled: false # # allows automatic connection fallback to TCP and TURN/TLS (if configured) when UDP has been unstable, default true # allow_tcp_fallback: true # # signaling RTT (in milliseconds) below which ICE/TCP is attempted on a UDP failure; at or above it, diff --git a/go.mod b/go.mod index 75b011646..45d8d8138 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095 - github.com/livekit/protocol v1.50.5-0.20260828115123-c6d0e234b41d + github.com/livekit/protocol v1.50.5-0.20260912174621-5d06e053f0e1 github.com/livekit/psrpc v0.7.5 github.com/mackerelio/go-osstat v0.2.8 github.com/magefile/mage v1.17.2 diff --git a/go.sum b/go.sum index 21843541e..6157c9ef5 100644 --- a/go.sum +++ b/go.sum @@ -164,8 +164,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095 h1:BcliKAXoMhl/nWmzQweQ5kmh4Qqagxl4s3Z5pvM/7AY= github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.50.5-0.20260828115123-c6d0e234b41d h1:qqObuGUWqdu1LbuexhZrrx2J7I25ZliwKRn/4ITu348= -github.com/livekit/protocol v1.50.5-0.20260828115123-c6d0e234b41d/go.mod h1:x7m1nX86XfvS0YoqXnVe0jixkCmAZglR4mcbRHURSro= +github.com/livekit/protocol v1.50.5-0.20260912174621-5d06e053f0e1 h1:b7gl3WgGApko8t1UZRC20INtcB7orifvqaTJgfn85c0= +github.com/livekit/protocol v1.50.5-0.20260912174621-5d06e053f0e1/go.mod h1:x7m1nX86XfvS0YoqXnVe0jixkCmAZglR4mcbRHURSro= github.com/livekit/psrpc v0.7.5 h1:WxfJIQ41X1b+48A1uzc8Gy9FhYEMxBJNCpCYqPdO/Ds= github.com/livekit/psrpc v0.7.5/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1 h1:fH+v4W+NFp9FfPzON6FaUFNmazGcctaAhb2P+Ksf+1s= diff --git a/pkg/config/config.go b/pkg/config/config.go index 575fa5345..44c486508 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -222,10 +222,12 @@ type CongestionControlConfig struct { SendSideBWE sendsidebwe.SendSideBWEConfig `yaml:"send_side_bwe,omitempty"` } -// FlexFECConfig controls FlexFEC-03 recovery on the publisher -> SFU leg. +// FlexFECConfig controls FlexFEC-03 independently on each media leg. type FlexFECConfig struct { // negotiate flexfec-03 with publishers and use it to recover lost upstream packets UpstreamEnabled bool `yaml:"upstream_enabled,omitempty"` + // allow repair packets for subscribers that negotiate flexfec-03 and select a per-track FEC preset + DownstreamEnabled bool `yaml:"downstream_enabled,omitempty"` } type PlayoutDelayConfig struct { diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index fac25df0f..aebc6b336 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -96,7 +96,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { PacketBufferSizeAudio: rtcConf.PacketBufferSizeAudio, }, Publisher: getPublisherConfig(false, flexFEC), - Subscriber: getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE), + Subscriber: getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE, flexFEC), flexFEC: flexFEC, }, nil } @@ -106,7 +106,7 @@ func (c *WebRTCConfig) UpdatePublisherConfig(consolidated bool) { } func (c *WebRTCConfig) UpdateSubscriberConfig(ccConf config.CongestionControlConfig) { - c.Subscriber = getSubscriberConfig(ccConf.UseSendSideBWEInterceptor || ccConf.UseSendSideBWE) + c.Subscriber = getSubscriberConfig(ccConf.UseSendSideBWEInterceptor || ccConf.UseSendSideBWE, c.flexFEC) } func (c *WebRTCConfig) SetBufferFactory(factory *buffer.Factory) { @@ -187,8 +187,9 @@ func getPublisherConfig(consolidated bool, flexFEC config.FlexFECConfig) Directi } } -func getSubscriberConfig(enableTWCC bool) DirectionConfig { +func getSubscriberConfig(enableTWCC bool, flexFEC config.FlexFECConfig) DirectionConfig { subscriberConfig := DirectionConfig{ + FlexFEC: FlexFECDirectionConfig{Enabled: flexFEC.DownstreamEnabled}, RTPHeaderExtension: RTPHeaderExtensionConfig{ Video: []string{ dd.ExtensionURI, diff --git a/pkg/rtc/signalling/signalhandler_fec_test.go b/pkg/rtc/signalling/signalhandler_fec_test.go new file mode 100644 index 000000000..5fe755150 --- /dev/null +++ b/pkg/rtc/signalling/signalhandler_fec_test.go @@ -0,0 +1,46 @@ +// 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 signalling + +import ( + "testing" + + "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +func TestSignalTrackFECProtection(t *testing.T) { + for _, level := range []*livekit.FECProtection{nil, livekit.FECProtection_FEC_NONE.Enum(), livekit.FECProtection_FEC_LOW.Enum(), livekit.FECProtection_FEC_MEDIUM.Enum(), livekit.FECProtection_FEC_HIGH.Enum()} { + participant := &typesfakes.FakeLocalParticipant{} + handler := NewSignalHandler(SignalHandlerParams{Participant: participant, Logger: logger.GetLogger()}) + signal := &livekit.SignalRequest{Message: &livekit.SignalRequest_TrackSetting{TrackSetting: &livekit.UpdateTrackSettings{ + TrackSids: []string{"track-a", "track-b"}, Fec: level, + }}} + wire, err := proto.Marshal(signal) + require.NoError(t, err) + decoded := &livekit.SignalRequest{} + require.NoError(t, proto.Unmarshal(wire, decoded)) + require.NoError(t, handler.HandleMessage(decoded)) + require.Equal(t, 2, participant.UpdateSubscribedTrackSettingsCallCount()) + for i, sid := range []livekit.TrackID{"track-a", "track-b"} { + actualSID, settings := participant.UpdateSubscribedTrackSettingsArgsForCall(i) + require.Equal(t, sid, actualSID) + require.Equal(t, level, settings.Fec) + } + } +} diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 27d18c382..54fa405f1 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -34,6 +34,7 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/livekit-server/pkg/telemetry/prometheus" ) const ( @@ -136,6 +137,8 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error) // Strip packet trailer if track has packet trailer but subscriber does not have cap stripPacketTrailer := params.MediaTrack.HasPacketTrailer() && !subSupportsPacketTrailer downTrack, err := sfu.NewDownTrack(sfu.DownTrackParams{ + EnableFlexFEC: params.SubscriberConfig.FlexFEC.Enabled, + OnFECSent: prometheus.RecordFECDownstream, Codecs: codecs, IsEncrypted: isEncrypted, Source: params.MediaTrack.Source(), @@ -316,6 +319,7 @@ func (t *SubscribedTrack) SetPublisherMuted(muted bool) { func (t *SubscribedTrack) UpdateSubscriberSettings(settings *livekit.UpdateTrackSettings, isImmediate bool) { t.settingsLock.Lock() + settings = mergeSubscriberSettings(t.settings, settings) if proto.Equal(t.settings, settings) { t.logger.Debugw("skipping subscriber track settings", "settings", logger.Proto(t.settings)) t.settingsLock.Unlock() @@ -323,7 +327,7 @@ func (t *SubscribedTrack) UpdateSubscriberSettings(settings *livekit.UpdateTrack } isImmediate = isImmediate || (!settings.Disabled && settings.Disabled != t.isMutedLocked()) - t.settings = utils.CloneProto(settings) + t.settings = settings t.logger.Debugw("saving subscriber track settings", "settings", logger.Proto(t.settings)) t.settingsLock.Unlock() @@ -375,6 +379,9 @@ func (t *SubscribedTrack) applySettings() { } t.logger.Debugw("applying subscriber track settings", "settings", logger.Proto(t.settings)) + if t.settings.Fec != nil { + dt.SetFECProtection(*t.settings.Fec) + } if t.settings.Disabled { dt.Mute(true) t.settingsLock.Unlock() diff --git a/pkg/rtc/subscribedtrack_fec.go b/pkg/rtc/subscribedtrack_fec.go new file mode 100644 index 000000000..bc4c2574b --- /dev/null +++ b/pkg/rtc/subscribedtrack_fec.go @@ -0,0 +1,41 @@ +// 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 rtc + +import ( + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/utils" +) + +// mergeSubscriberSettings retains an explicitly requested FEC level across +// legacy clients' quality/mute updates, which omit the new optional field. Clone +// both the incoming message and inherited value so cached settings are immutable. +func mergeSubscriberSettings(previous, next *livekit.UpdateTrackSettings) *livekit.UpdateTrackSettings { + settings := utils.CloneProto(next) + fec := settings.Fec + if fec == nil && previous != nil { + fec = previous.Fec + } + if fec != nil { + level := *fec + switch level { + case livekit.FECProtection_FEC_NONE, livekit.FECProtection_FEC_LOW, livekit.FECProtection_FEC_MEDIUM, livekit.FECProtection_FEC_HIGH: + default: + level = livekit.FECProtection_FEC_NONE + } + settings.Fec = level.Enum() + } + return settings +} diff --git a/pkg/rtc/subscribedtrack_fec_test.go b/pkg/rtc/subscribedtrack_fec_test.go new file mode 100644 index 000000000..38b3c78cc --- /dev/null +++ b/pkg/rtc/subscribedtrack_fec_test.go @@ -0,0 +1,146 @@ +// 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 rtc + +import ( + "testing" + + "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" + "github.com/livekit/livekit-server/pkg/sfu" + "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/livekit-server/pkg/sfu/sfufakes" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" +) + +func TestSubscriberFECSettings(t *testing.T) { + require.Nil(t, mergeSubscriberSettings(nil, &livekit.UpdateTrackSettings{}).Fec) + for _, tc := range []struct { + requested, want livekit.FECProtection + }{ + {livekit.FECProtection_FEC_NONE, livekit.FECProtection_FEC_NONE}, + {livekit.FECProtection_FEC_LOW, livekit.FECProtection_FEC_LOW}, + {livekit.FECProtection_FEC_MEDIUM, livekit.FECProtection_FEC_MEDIUM}, + {livekit.FECProtection_FEC_HIGH, livekit.FECProtection_FEC_HIGH}, + {-1, livekit.FECProtection_FEC_NONE}, + {99, livekit.FECProtection_FEC_NONE}, + } { + requested := &livekit.UpdateTrackSettings{Fec: tc.requested.Enum()} + cached := mergeSubscriberSettings(nil, requested) + require.NotNil(t, cached.Fec) + require.Equal(t, tc.want, *cached.Fec) + require.Equal(t, tc.requested, *requested.Fec, "do not change caller-owned messages") + legacyUpdate := &livekit.UpdateTrackSettings{Disabled: true, Width: 640} + merged := mergeSubscriberSettings(cached, legacyUpdate) + require.True(t, merged.Disabled) + require.EqualValues(t, 640, merged.Width) + require.Equal(t, cached.Fec, merged.Fec) + require.Nil(t, legacyUpdate.Fec) + *merged.Fec = 42 + require.Equal(t, tc.want, *cached.Fec, "cached option must not alias later settings") + } +} + +func TestSubscriberFECPersistsAcrossResubscription(t *testing.T) { + sub := newMediaTrackSubscription("subscriber", "track", logger.GetLogger()) + sub.setSettings(&livekit.UpdateTrackSettings{Fec: livekit.FECProtection_FEC_MEDIUM.Enum()}) + first := &typesfakes.FakeSubscribedTrack{} + sub.setSubscribedTrack(first) + got, immediate := first.UpdateSubscriberSettingsArgsForCall(0) + require.Equal(t, livekit.FECProtection_FEC_MEDIUM, *got.Fec) + require.True(t, immediate) + sub.setSettings(&livekit.UpdateTrackSettings{Width: 640, Height: 360}) + got, immediate = first.UpdateSubscriberSettingsArgsForCall(1) + require.Equal(t, livekit.FECProtection_FEC_MEDIUM, *got.Fec) + require.False(t, immediate) + sub.setSettings(&livekit.UpdateTrackSettings{Fec: livekit.FECProtection_FEC_NONE.Enum()}) + sub.setSettings(&livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH}) + sub.setSubscribedTrack(nil) + second := &typesfakes.FakeSubscribedTrack{} + sub.setSubscribedTrack(second) + got, immediate = second.UpdateSubscriberSettingsArgsForCall(0) + require.NotNil(t, got.Fec) + require.Zero(t, *got.Fec, "explicit off must survive resubscription and quality updates") + require.True(t, immediate) +} + +func TestSubscriberFECAppliesToDownTrack(t *testing.T) { + codec := webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000}, + PayloadType: 96, + } + receiver := &sfufakes.FakeTrackReceiver{} + receiver.TrackIDReturns("track") + receiver.CodecReturns(codec) + receiver.AddOnReadyCalls(func(f func()) { f() }) + receiver.GetLayeredBitrateReturns([]int32{0}, sfu.Bitrates{{100_000}}) + dt, err := sfu.NewDownTrack(sfu.DownTrackParams{ + EnableFlexFEC: true, + Codecs: []webrtc.RTPCodecParameters{codec}, + Receiver: receiver, + BufferFactory: buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory(), + MaxTrack: 500, + Logger: logger.GetLogger(), + Listener: &sfufakes.FakeDownTrackListener{}, + }) + require.NoError(t, err) + t.Cleanup(func() { dt.CloseWithFlush(false, true) }) + _, err = dt.Bind(subscriberFECTrackContext{codecs: []webrtc.RTPCodecParameters{ + codec, + {RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeFlexFEC03, ClockRate: 90000}, PayloadType: 115}, + }}) + require.NoError(t, err) + dt.UpTrackMaxPublishedLayerChange(0) + dt.UpTrackMaxTemporalLayerSeenChange(0) + mediaTrack := &typesfakes.FakeMediaTrack{} + mediaTrack.ToProtoReturns(&livekit.TrackInfo{Type: livekit.TrackType_VIDEO}) + sub := &SubscribedTrack{ + params: SubscribedTrackParams{MediaTrack: mediaTrack}, + downTrack: dt, + logger: logger.GetLogger(), + versionGenerator: utils.NewDefaultTimedVersionGenerator(), + } + for _, tc := range []struct { + level *livekit.FECProtection + want int64 + }{ + {nil, 100_000}, + {livekit.FECProtection_FEC_MEDIUM.Enum(), 125_000}, + {nil, 125_000}, + {livekit.FECProtection_FEC_NONE.Enum(), 100_000}, + {nil, 100_000}, + {livekit.FECProtection_FEC_LOW.Enum(), 115_000}, + {livekit.FECProtection_FEC_HIGH.Enum(), 135_000}, + {livekit.FECProtection(99).Enum(), 100_000}, + } { + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH, Fec: tc.level}, true) + dt.AllocateOptimal(false, false) + require.Equal(t, tc.want, dt.BandwidthRequested(), "client settings must reach the negotiated encoder and allocator") + } +} + +type subscriberFECTrackContext struct { + webrtc.TrackLocalContext + codecs []webrtc.RTPCodecParameters +} + +func (c subscriberFECTrackContext) CodecParameters() []webrtc.RTPCodecParameters { return c.codecs } +func (subscriberFECTrackContext) SSRC() webrtc.SSRC { return 123 } +func (subscriberFECTrackContext) SSRCRetransmission() webrtc.SSRC { return 0 } +func (subscriberFECTrackContext) SSRCForwardErrorCorrection() webrtc.SSRC { return 456 } +func (subscriberFECTrackContext) WriteStream() webrtc.TrackLocalWriter { return nil } diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index 3b5ddd849..5e4567d74 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/pkg/rtc/subscriptionmanager.go @@ -1479,6 +1479,7 @@ func (s *mediaTrackSubscription) getKind() (livekit.TrackType, bool) { func (s *mediaTrackSubscription) setSettings(settings *livekit.UpdateTrackSettings) { s.lock.Lock() + settings = mergeSubscriberSettings(s.settings, settings) s.settings = settings subTrack := s.subscribedTrack s.lock.Unlock() diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 5bf75c9b8..1659da944 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -338,6 +338,11 @@ func newPeerConnection( onBandwidthEstimator func(estimator cc.BandwidthEstimator), ) (*webrtc.PeerConnection, *webrtc.MediaEngine, *sfuinterceptor.RTXInfoExtractorFactory, error) { directionConfig := params.DirectionConfig + if params.IsSendSide { + // A single publisher PC can also carry subscriptions. Register the + // union here; recovery and generation remain independently gated. + directionConfig.FlexFEC.Enabled = directionConfig.FlexFEC.Enabled || params.Config.Subscriber.FlexFEC.Enabled + } if params.AllowPlayoutDelay { directionConfig.RTPHeaderExtension.Video = append(directionConfig.RTPHeaderExtension.Video, pd.PlayoutDelayURI) } @@ -1024,7 +1029,7 @@ func (t *PCTransport) queueOrConfigureSender( filterOutH264HighProfile: !t.params.IsOfferer, enableAudioStereo: enableAudioStereo, enableAudioNACK: enableAudioNACK, - keepFlexFEC: t.params.DirectionConfig.FlexFEC.Enabled, + keepFlexFEC: t.params.Config.Subscriber.FlexFEC.Enabled, } if !t.params.IsOfferer { t.sendersPendingConfigMu.Lock() @@ -3282,12 +3287,19 @@ func (t *PCTransport) restrictReceiverCodecsToPublishList() { if receiver == nil { continue } + keepFlexFEC := t.params.DirectionConfig.FlexFEC.Enabled + if tr.Direction() == webrtc.RTPTransceiverDirectionSendrecv && t.params.Config.Subscriber.FlexFEC.Enabled { + // Both directions share the codec list on a sendrecv m-section. + // Retain repair support for its subscription; upstream recovery + // is still gated independently when processing FEC-FR pairs. + keepFlexFEC = true + } filtered := filterCodecs( receiver.GetParameters().Codecs, t.params.EnabledPublishCodecs, t.params.DirectionConfig.RTCPFeedback, false, - t.params.DirectionConfig.FlexFEC.Enabled, + keepFlexFEC, ) if len(filtered) == 0 { continue diff --git a/pkg/rtc/transport_fec_test.go b/pkg/rtc/transport_fec_test.go index e52ae01a5..a9f530657 100644 --- a/pkg/rtc/transport_fec_test.go +++ b/pkg/rtc/transport_fec_test.go @@ -23,6 +23,7 @@ import ( "github.com/stretchr/testify/require" "github.com/livekit/livekit-server/pkg/config" + "github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes" protoCodecs "github.com/livekit/protocol/codecs" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" @@ -203,3 +204,54 @@ func TestWebRTCConfigFlexFEC(t *testing.T) { assert.True(t, webRTCConfig.Publisher.FlexFEC.Enabled) }) } + +func TestWebRTCConfigDownstreamFlexFEC(t *testing.T) { + for _, upstream := range []bool{false, true} { + for _, downstream := range []bool{false, true} { + conf, err := config.NewConfig("", true, nil, nil) + require.NoError(t, err) + conf.RTC.TCPPort = 0 + conf.RTC.FlexFEC = config.FlexFECConfig{UpstreamEnabled: upstream, DownstreamEnabled: downstream} + c, err := NewWebRTCConfig(conf) + require.NoError(t, err) + for _, twcc := range []bool{false, true} { + c.UpdateSubscriberConfig(config.CongestionControlConfig{UseSendSideBWE: twcc}) + c.UpdatePublisherConfig(true) + require.Equal(t, upstream, c.Publisher.FlexFEC.Enabled) + require.Equal(t, downstream, c.Subscriber.FlexFEC.Enabled) + } + } + } +} + +func TestSinglePCDownstreamFlexFEC(t *testing.T) { + codecs := []*livekit.Codec{{Mime: "video/VP8"}, {Mime: "video/rtx"}} + for _, direction := range []webrtc.RTPTransceiverDirection{webrtc.RTPTransceiverDirectionSendonly, webrtc.RTPTransceiverDirectionSendrecv, webrtc.RTPTransceiverDirectionRecvonly} { + t.Run(direction.String(), func(t *testing.T) { + conf := &WebRTCConfig{Subscriber: DirectionConfig{FlexFEC: FlexFECDirectionConfig{Enabled: true}}} + server, err := NewPCTransport(TransportParams{ + Config: conf, IsSendSide: true, Handler: &transportfakes.FakeHandler{}, + EnabledPublishCodecs: codecs, EnabledSubscribeCodecs: codecs, + }) + require.NoError(t, err) + defer server.Close() + tr, err := server.pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{Direction: direction}) + require.NoError(t, err) + if direction != webrtc.RTPTransceiverDirectionRecvonly { + server.queueOrConfigureSender(tr, codecs, RTCPFeedbackConfig{}, false, true) + require.Len(t, server.sendersPendingConfig, 1) + require.True(t, server.sendersPendingConfig[0].keepFlexFEC) + configureSender(server.sendersPendingConfig[0], nil) + } + server.restrictReceiverCodecsToPublishList() + offer, err := server.pc.CreateOffer(nil) + require.NoError(t, err) + if direction == webrtc.RTPTransceiverDirectionRecvonly { + require.NotContains(t, offer.SDP, "flexfec-03") + } else { + require.Contains(t, offer.SDP, "flexfec-03/90000") + require.Contains(t, offer.SDP, "ssrc-group:FEC-FR") + } + }) + } +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 1b993158c..a90fb2397 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -42,6 +42,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/bwe" "github.com/livekit/livekit-server/pkg/sfu/ccutils" "github.com/livekit/livekit-server/pkg/sfu/connectionquality" + "github.com/livekit/livekit-server/pkg/sfu/flexfec" "github.com/livekit/livekit-server/pkg/sfu/pacer" "github.com/livekit/livekit-server/pkg/sfu/packettrailer" act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime" @@ -226,6 +227,8 @@ var _ TrackSender = (*DownTrack)(nil) type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) type DownTrackParams struct { + EnableFlexFEC bool + OnFECSent func(packets int, bytes int) Codecs []webrtc.RTPCodecParameters IsEncrypted bool Source livekit.TrackSource @@ -255,15 +258,18 @@ type DownTrackParams struct { // - closed // once closed, a DownTrack cannot be re-used. type DownTrack struct { - params DownTrackParams - id livekit.TrackID - kind webrtc.RTPCodecType - ssrc uint32 - ssrcRTX uint32 - payloadType atomic.Uint32 - payloadTypeRTX atomic.Uint32 - sequencer *sequencer - rtxSequenceNumber atomic.Uint64 + fecEncoder atomic.Pointer[flexfec.Encoder] + fecProtectionPercent atomic.Uint32 + fecLock sync.Mutex + params DownTrackParams + id livekit.TrackID + kind webrtc.RTPCodecType + ssrc uint32 + ssrcRTX uint32 + payloadType atomic.Uint32 + payloadTypeRTX atomic.Uint32 + sequencer *sequencer + rtxSequenceNumber atomic.Uint64 receiverLock sync.RWMutex receiver TrackReceiver @@ -376,6 +382,7 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) { createdAt: time.Now().UnixNano(), receiver: params.Receiver, } + d.fecProtectionPercent.Store(flexfec.DefaultProtectionPercent) d.lastUnmutedAt.Store(time.Now()) d.codec.Store(codec) @@ -575,6 +582,7 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, ) d.params.Logger.Debugw("DownTrack.Bind", logFields...) + d.bindFEC(t) d.writeStream = t.WriteStream() if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrc).(*buffer.RTCPReader); rr != nil { rr.OnPacket(func(pkt []byte) { @@ -725,6 +733,7 @@ func (d *DownTrack) handleUpstreamCodecChange(mimeType string) { func (d *DownTrack) Unbind(_ webrtc.TrackLocalContext) error { d.bindLock.Lock() d.setBindStateLocked(bindStateUnbound) + d.closeFEC() d.bindLock.Unlock() return nil } @@ -1154,6 +1163,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { HeaderPool: RTPHeaderFactory, HeaderSize: headerSize, Payload: payload, + FEC: d.fecEncoder.Load(), ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()), AbsSendTimeExtID: uint8(d.absSendTimeExtID), TransportWideExtID: uint8(d.transportWideExtID), @@ -1466,6 +1476,7 @@ func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) { d.rtpStats.Stop() d.rtpStatsRTX.Stop() + d.closeFEC() d.params.Logger.Debugw( "rtp stats", "direction", "downstream", @@ -1640,17 +1651,17 @@ func (d *DownTrack) IsDeficient() bool { } func (d *DownTrack) BandwidthRequested() int64 { - _, brs := d.Receiver().GetLayeredBitrate() + _, brs := d.getLayeredBitrateWithFEC() return d.forwarder.BandwidthRequested(brs) } func (d *DownTrack) DistanceToDesired() float64 { - al, brs := d.Receiver().GetLayeredBitrate() + al, brs := d.getLayeredBitrateWithFEC() return d.forwarder.DistanceToDesired(al, brs) } func (d *DownTrack) AllocateOptimal(allowOvershoot bool, hold bool) VideoAllocation { - al, brs := d.Receiver().GetLayeredBitrate() + al, brs := d.getLayeredBitrateWithFEC() allocation := d.forwarder.AllocateOptimal(al, brs, allowOvershoot, hold) d.postKeyFrameRequestEvent() d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason) @@ -1658,7 +1669,7 @@ func (d *DownTrack) AllocateOptimal(allowOvershoot bool, hold bool) VideoAllocat } func (d *DownTrack) ProvisionalAllocatePrepare() { - al, brs := d.Receiver().GetLayeredBitrate() + al, brs := d.getLayeredBitrateWithFEC() d.forwarder.ProvisionalAllocatePrepare(al, brs) } @@ -1700,7 +1711,7 @@ func (d *DownTrack) ProvisionalAllocateCommit() VideoAllocation { } func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64, allowOvershoot bool) (VideoAllocation, bool) { - al, brs := d.Receiver().GetLayeredBitrate() + al, brs := d.getLayeredBitrateWithFEC() allocation, available := d.forwarder.AllocateNextHigher(availableChannelCapacity, al, brs, allowOvershoot) d.postKeyFrameRequestEvent() d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason) @@ -1708,7 +1719,7 @@ func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64, allowOver } func (d *DownTrack) GetNextHigherTransition(allowOvershoot bool) (VideoTransition, bool) { - availableLayers, brs := d.Receiver().GetLayeredBitrate() + availableLayers, brs := d.getLayeredBitrateWithFEC() transition, available := d.forwarder.GetNextHigherTransition(brs, allowOvershoot) d.params.Logger.Debugw( "stream: get next higher layer", @@ -1721,7 +1732,7 @@ func (d *DownTrack) GetNextHigherTransition(allowOvershoot bool) (VideoTransitio } func (d *DownTrack) Pause() VideoAllocation { - al, brs := d.Receiver().GetLayeredBitrate() + al, brs := d.getLayeredBitrateWithFEC() allocation := d.forwarder.Pause(al, brs) d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason) return allocation diff --git a/pkg/sfu/downtrack_fec.go b/pkg/sfu/downtrack_fec.go new file mode 100644 index 000000000..9d71ae65c --- /dev/null +++ b/pkg/sfu/downtrack_fec.go @@ -0,0 +1,101 @@ +// 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 sfu + +import ( + "strings" + + "github.com/livekit/protocol/livekit" + "github.com/pion/webrtc/v4" + + "github.com/livekit/livekit-server/pkg/sfu/flexfec" +) + +// bindFEC runs under bindLock, using the complete negotiated codec list saved +// before Pion narrows the TrackLocalContext to the selected media codec. +func (d *DownTrack) bindFEC(t webrtc.TrackLocalContext) { + d.closeFEC() + d.fecLock.Lock() + defer d.fecLock.Unlock() + if !d.params.EnableFlexFEC || d.kind != webrtc.RTPCodecTypeVideo || t.SSRCForwardErrorCorrection() == 0 { + return + } + for _, codec := range d.negotiatedCodecParameters { + if strings.EqualFold(codec.MimeType, webrtc.MimeTypeFlexFEC03) { + encoder := flexfec.NewEncoder(uint8(codec.PayloadType), uint32(t.SSRCForwardErrorCorrection()), d.params.OnFECSent) + encoder.SetProtectionPercent(d.fecProtectionPercent.Load()) + d.fecEncoder.Store(encoder) + return + } + } +} + +func (d *DownTrack) closeFEC() { + d.fecLock.Lock() + defer d.fecLock.Unlock() + if encoder := d.fecEncoder.Swap(nil); encoder != nil { + encoder.Close() + } +} + +// Include the nominal repair overhead in every allocator input, so layer +// selection leaves room for repair traffic. The pacer/BWE account actual bytes. +func (d *DownTrack) getLayeredBitrateWithFEC() ([]int32, Bitrates) { + layers, bitrates := d.Receiver().GetLayeredBitrate() + if d.fecEncoder.Load() != nil { + percent := int64(d.fecProtectionPercent.Load()) + if percent == 0 { + return layers, bitrates + } + for spatial := range bitrates { + for temporal, bitrate := range bitrates[spatial] { + bitrates[spatial][temporal] += bitrate * percent / 100 + } + } + } + return layers, bitrates +} + +// SetFECProtection updates a subscriber's video FEC preset, using the same +// 0/15/25/35 percent levels as the publish-track options. Unknown levels disable FEC. +// It can run before Bind and never enables FEC unless it was negotiated and +// permitted by the server. Changes take effect without SDP renegotiation. +func (d *DownTrack) SetFECProtection(level livekit.FECProtection) { + var percent uint32 + switch level { + case livekit.FECProtection_FEC_LOW: + percent = 15 + case livekit.FECProtection_FEC_MEDIUM: + percent = 25 + case livekit.FECProtection_FEC_HIGH: + percent = 35 + } + d.fecLock.Lock() + previous := d.fecProtectionPercent.Swap(percent) + if previous == percent { + d.fecLock.Unlock() + return + } + encoder := d.fecEncoder.Load() + if encoder != nil { + encoder.SetProtectionPercent(percent) + } + d.fecLock.Unlock() + if encoder != nil { + if listener := d.getStreamAllocatorListener(); listener != nil { + listener.OnSubscriptionChanged(d) + } + } +} diff --git a/pkg/sfu/downtrack_fec_integration_test.go b/pkg/sfu/downtrack_fec_integration_test.go new file mode 100644 index 000000000..e8b550a6e --- /dev/null +++ b/pkg/sfu/downtrack_fec_integration_test.go @@ -0,0 +1,253 @@ +// 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 sfu_test + +import ( + "errors" + "fmt" + "io" + "sync/atomic" + "testing" + "time" + + "github.com/pion/interceptor" + "github.com/pion/rtp" + "github.com/pion/transport/v4/packetio" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/sfu" + "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/livekit-server/pkg/sfu/flexfec" + "github.com/livekit/livekit-server/pkg/sfu/pacer" + "github.com/livekit/livekit-server/pkg/sfu/sfufakes" + "github.com/livekit/livekit-server/pkg/sfu/testutils" + "github.com/livekit/livekit-server/pkg/testutils/vnettest" + "github.com/livekit/mediatransportutil/pkg/codec" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +// Capture decrypted RTP at the subscriber before Pion demultiplexes repair SSRCs. +// Retain independent packets because SRTP reuses its input buffer. +type fecWireCapture struct { + io.ReadWriteCloser + capture *packetCapture +} + +func (c *fecWireCapture) Write(raw []byte) (int, error) { + p := &rtp.Packet{} + if err := p.Unmarshal(raw); err == nil { + c.capture.add(p.Clone()) + } + return c.ReadWriteCloser.Write(raw) +} + +func TestDownTrackFlexFECOnWire(t *testing.T) { + for _, tc := range []struct { + name string + enabled, negotiated, encrypted bool + pacing string + level *livekit.FECProtection + protectionPercent uint32 + }{ + {"pass through", true, true, false, "pass", livekit.FECProtection_FEC_MEDIUM.Enum(), 25}, + {"queued", true, true, false, "queue", livekit.FECProtection_FEC_MEDIUM.Enum(), 25}, + {"low", true, true, false, "queue", livekit.FECProtection_FEC_LOW.Enum(), 15}, + {"high", true, true, false, "leaky", livekit.FECProtection_FEC_HIGH.Enum(), 35}, + {"none", true, true, false, "pass", livekit.FECProtection_FEC_NONE.Enum(), 0}, + {"default none", true, true, false, "pass", nil, 0}, + {"paced encrypted", true, true, true, "leaky", livekit.FECProtection_FEC_MEDIUM.Enum(), 25}, + {"subscriber declines", true, false, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35}, + {"disabled", false, true, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35}, + } { + t.Run(tc.name, func(t *testing.T) { + h := vnettest.NewHosts(t) + factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() + capture := &packetCapture{} + makePC := func(sender bool, enableFEC bool) *webrtc.PeerConnection { + me := vnettest.NewMediaEngine(t, vnettest.MediaEngineConfig{Video: true}) + if enableFEC { + require.NoError(t, me.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeFlexFEC03, ClockRate: 90000, SDPFmtpLine: "repair-window=10000000"}, + PayloadType: 118, + }, webrtc.RTPCodecTypeVideo)) + } + net := h.AnswerNet + if sender { + net = h.OfferNet + } + se := vnettest.NewSettingEngine(net) + if sender { + se.BufferFactory = factory.GetOrNew + } else { + se.BufferFactory = func(kind packetio.BufferPacketType, _ uint32) io.ReadWriteCloser { + b := packetio.NewBuffer() + if kind == packetio.RTPBufferPacket { + return &fecWireCapture{ReadWriteCloser: b, capture: capture} + } + return b + } + } + pc, err := webrtc.NewAPI(webrtc.WithMediaEngine(me), webrtc.WithSettingEngine(se), webrtc.WithInterceptorRegistry(&interceptor.Registry{})).NewPeerConnection(webrtc.Configuration{}) + require.NoError(t, err) + t.Cleanup(func() { _ = pc.Close() }) + return pc + } + sender, sub := makePC(true, tc.enabled), makePC(false, tc.negotiated) + captureTrack(sub) // drain media normally, even when FEC is declined + var p pacer.Pacer + switch tc.pacing { + case "queue": + p = pacer.NewNoQueue(logger.GetLogger(), newNullBWE()) + case "leaky": + p = pacer.NewLeakyBucket(logger.GetLogger(), newNullBWE(), time.Millisecond, 10_000_000) + default: + p = pacer.NewPassThrough(logger.GetLogger(), newNullBWE()) + } + t.Cleanup(p.Stop) + var repairsSent atomic.Int32 + dt, err := sfu.NewDownTrack(sfu.DownTrackParams{ + Codecs: []webrtc.RTPCodecParameters{vp8CodecParams}, + Receiver: newFakeTrackReceiver(vp8CodecParams), BufferFactory: factory, + Pacer: p, Logger: logger.GetLogger(), Listener: &sfufakes.FakeDownTrackListener{}, + StreamID: "fec-stream", SubID: "fec-sub", MaxTrack: 500, + EnableFlexFEC: tc.enabled, IsEncrypted: tc.encrypted, + OnFECSent: func(n int, _ int) { repairsSent.Add(int32(n)) }, + }) + require.NoError(t, err) + if tc.level != nil { + dt.SetFECProtection(*tc.level) + } + tr, err := sender.AddTransceiverFromTrack(dt, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly}) + require.NoError(t, err) + dt.SetTransceiver(tr) + vnettest.SignalPair(t, sender, sub) + dt.SetConnected() + require.Eventually(t, dt.IsWritableForTest, 10*time.Second, 10*time.Millisecond) + dt.ForceForwardLayerForTest(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + t.Cleanup(func() { dt.CloseWithFlush(false, true) }) + + negotiated := tc.enabled && tc.negotiated + fecSSRC := uint32(tr.Sender().GetParameters().Encodings[0].FEC.SSRC) + if negotiated { + require.NotZero(t, fecSSRC) + require.Contains(t, sender.LocalDescription().SDP, fmt.Sprintf("a=ssrc-group:FEC-FR %d %d", dt.SSRC(), fecSSRC)) + require.Contains(t, sub.LocalDescription().SDP, "flexfec-03/90000") + } else { + require.Zero(t, fecSSRC) + } + + const mediaCount = 20 + sendMedia := func(start, count int) { + for i := start; i < start+count; i++ { + ep, err := testutils.GetTestExtPacketVP8(&testutils.TestExtPacketParams{ + SequenceNumber: uint16(6000 + i), Timestamp: uint32(270000 + i*3000), SSRC: 0x44444444, + PayloadType: 96, PayloadSize: 100 + i, IsKeyFrame: true, Marker: true, + }, &codec.VP8{FirstByte: 0x10, S: true, PictureID: uint16(i + 1), IsKeyFrame: true}) + require.NoError(t, err) + ep.Packet.Payload = distinctivePayload(byte(i), 100+i) + require.EqualValues(t, 1, dt.WriteRTP(ep, 0)) + clear(ep.Packet.Payload) // forwarding and encoding must own retained bytes + } + } + sendMedia(0, mediaCount) + expected := mediaCount + if negotiated { + expected += mediaCount * int(tc.protectionPercent) / 100 + } + require.Eventually(t, func() bool { return capture.count() >= expected }, 10*time.Second, 10*time.Millisecond) + require.Equal(t, expected, capture.count()) + packets := capture.all() + var media, repair []*rtp.Packet + for _, packet := range packets { + if packet.SSRC == dt.SSRC() { + media = append(media, packet) + } else { + require.Equal(t, fecSSRC, packet.SSRC) + require.EqualValues(t, 118, packet.PayloadType) + repair = append(repair, packet) + } + } + require.Len(t, media, mediaCount) + require.EqualValues(t, len(repair), repairsSent.Load()) + if !negotiated || tc.protectionPercent == 0 { + require.Empty(t, repair) + return + } + require.Len(t, repair, mediaCount*int(tc.protectionPercent)/100) + for i := 1; i < len(repair); i++ { + require.Equal(t, repair[i-1].SequenceNumber+1, repair[i].SequenceNumber) + } + // Drop media in the second and third groups (protected by every enabled + // preset) and recover from the actual wire + // representation, including translated SSRC/PT/sequence and payload. + stored := map[uint16][]byte{} + for i, packet := range media { + if i != 7 && i != 12 { + stored[packet.SequenceNumber], _ = packet.Marshal() + } + } + decoder := flexfec.NewDecoder(fecSSRC, dt.SSRC(), func(sn uint16, dst []byte) (int, error) { + raw, ok := stored[sn] + if !ok { + return 0, errors.New("dropped media") + } + return copy(dst, raw), nil + }, logger.GetLogger()) + recoveredPackets := map[uint16][]byte{} + for _, packet := range repair { + for _, recovered := range decoder.DecodeFEC(packet) { + raw, err := recovered.Marshal() + require.NoError(t, err) + recoveredPackets[recovered.SequenceNumber] = raw + stored[recovered.SequenceNumber] = raw + } + } + require.Len(t, recoveredPackets, 2) + for _, lost := range []*rtp.Packet{media[7], media[12]} { + want, _ := lost.Marshal() + require.Equal(t, want, recoveredPackets[lost.SequenceNumber]) + } + if tc.name == "pass through" { + // Updating protection keeps the same negotiated repair stream. + sequence := repair[len(repair)-1].SequenceNumber + start := mediaCount + for _, update := range []struct { + level livekit.FECProtection + percent int + }{ + {livekit.FECProtection_FEC_HIGH, 35}, + {livekit.FECProtection_FEC_NONE, 0}, + {livekit.FECProtection_FEC_LOW, 15}, + } { + before := capture.count() + dt.SetFECProtection(update.level) + sendMedia(start, mediaCount) + start += mediaCount + numRepair := mediaCount * update.percent / 100 + require.Eventually(t, func() bool { return capture.count() >= before+mediaCount+numRepair }, 5*time.Second, 10*time.Millisecond) + require.Equal(t, before+mediaCount+numRepair, capture.count()) + for _, packet := range capture.all()[before:] { + if packet.SSRC == fecSSRC { + require.Equal(t, sequence+1, packet.SequenceNumber) + sequence = packet.SequenceNumber + } + } + } + } + }) + } +} diff --git a/pkg/sfu/downtrack_fec_test.go b/pkg/sfu/downtrack_fec_test.go new file mode 100644 index 000000000..651c9fb01 --- /dev/null +++ b/pkg/sfu/downtrack_fec_test.go @@ -0,0 +1,146 @@ +// 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 sfu + +import ( + "testing" + + "github.com/livekit/livekit-server/pkg/sfu/flexfec" + "github.com/livekit/protocol/livekit" + "github.com/pion/rtp" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" +) + +type fecTrackContext struct { + webrtc.TrackLocalContext + ssrcFEC webrtc.SSRC +} + +func (c fecTrackContext) SSRCForwardErrorCorrection() webrtc.SSRC { return c.ssrcFEC } + +func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) { + fecCodec := webrtc.RTPCodecParameters{RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: "video/FLEXFEC-03", ClockRate: 90000}, PayloadType: 118} + for _, name := range []string{"enabled", "disabled", "audio", "no codec", "no ssrc"} { + t.Run(name, func(t *testing.T) { + d := &DownTrack{params: DownTrackParams{EnableFlexFEC: name != "disabled"}, kind: webrtc.RTPCodecTypeVideo, negotiatedCodecParameters: []webrtc.RTPCodecParameters{fecCodec}} + c := fecTrackContext{ssrcFEC: 456} + switch name { + case "audio": + d.kind = webrtc.RTPCodecTypeAudio + case "no codec": + d.negotiatedCodecParameters = nil + case "no ssrc": + c.ssrcFEC = 0 + } + d.SetFECProtection(livekit.FECProtection_FEC_MEDIUM) + d.bindFEC(c) + if name != "enabled" { + require.Nil(t, d.fecEncoder.Load()) + return + } + old := d.fecEncoder.Load() + require.NotNil(t, old) + p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96} + for i := range flexfec.MediaPacketsPerGroup { + p.SequenceNumber++ + repair := old.Encode(&p, []byte{1, 2, 3}) + if i == flexfec.MediaPacketsPerGroup-1 { + require.Len(t, repair, 1) + require.EqualValues(t, 456, repair[0].SSRC) + require.EqualValues(t, 118, repair[0].PayloadType, "use negotiated PT") + } + } + d.bindFEC(c) + require.NotSame(t, old, d.fecEncoder.Load()) + for range flexfec.MediaPacketsPerGroup { + p.SequenceNumber++ + require.Empty(t, old.Encode(&p, []byte{1, 2, 3}), "old queued writes cannot generate repair") + } + d.closeFEC() + require.Nil(t, d.fecEncoder.Load()) + }) + } +} + +func TestDownTrackFECBandwidthReservation(t *testing.T) { + rates := Bitrates{{100_000, 200_000}, {300_000, 600_000}} + receiver := &fecTrackReceiver{rates: rates} + d := &DownTrack{receiver: receiver} + _, got := d.getLayeredBitrateWithFEC() + require.Equal(t, rates, got) + d.SetFECProtection(livekit.FECProtection_FEC_MEDIUM) + d.fecEncoder.Store(flexfec.NewEncoder(115, 456, nil)) + layers, got := d.getLayeredBitrateWithFEC() + require.Equal(t, []int32{0, 1}, layers) + require.EqualValues(t, 125_000, got[0][0]) + require.EqualValues(t, 750_000, got[1][1]) + _, unchanged := receiver.GetLayeredBitrate() + require.Equal(t, rates, unchanged, "do not mutate shared upstream bitrate estimates") +} + +type fecTrackReceiver struct { + TrackReceiver + rates Bitrates +} + +func (r *fecTrackReceiver) GetLayeredBitrate() ([]int32, Bitrates) { return []int32{0, 1}, r.rates } + +func TestDownTrackFECProtectionChanges(t *testing.T) { + d := &DownTrack{ + params: DownTrackParams{EnableFlexFEC: true}, kind: webrtc.RTPCodecTypeVideo, + receiver: &fecTrackReceiver{rates: Bitrates{{100_000}}}, + negotiatedCodecParameters: []webrtc.RTPCodecParameters{{RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeFlexFEC03}, PayloadType: 115}}, + } + // Cached settings can arrive before Pion binds the track. + d.SetFECProtection(livekit.FECProtection_FEC_MEDIUM) + d.bindFEC(fecTrackContext{ssrcFEC: 456}) + encoder := d.fecEncoder.Load() + require.NotNil(t, encoder) + _, rates := d.getLayeredBitrateWithFEC() + require.EqualValues(t, 125_000, rates[0][0]) + p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96} + send := func() int { + count := 0 + for range 20 { + p.SequenceNumber++ + count += len(d.fecEncoder.Load().Encode(&p, []byte{1, 2, 3})) + } + return count + } + require.Equal(t, 5, send()) + for _, tc := range []struct { + level livekit.FECProtection + percent int64 + }{ + {livekit.FECProtection_FEC_NONE, 0}, + {livekit.FECProtection_FEC_LOW, 15}, + {livekit.FECProtection_FEC_MEDIUM, 25}, + {livekit.FECProtection_FEC_HIGH, 35}, + {-1, 0}, + {99, 0}, + } { + d.SetFECProtection(tc.level) + require.Same(t, encoder, d.fecEncoder.Load(), "updates do not rebind or reset repair sequencing") + _, rates = d.getLayeredBitrateWithFEC() + require.EqualValues(t, 100_000+1000*tc.percent, rates[0][0]) + require.EqualValues(t, tc.percent/5, send()) + } + d.closeFEC() + d.SetFECProtection(livekit.FECProtection_FEC_HIGH) + d.bindFEC(fecTrackContext{ssrcFEC: 456}) + require.NotSame(t, encoder, d.fecEncoder.Load()) + require.Equal(t, 7, send(), "rebind retains the subscriber's preset") +} diff --git a/pkg/sfu/flexfec/decoder.go b/pkg/sfu/flexfec/decoder.go index 95e0d25dc..21870d8ca 100644 --- a/pkg/sfu/flexfec/decoder.go +++ b/pkg/sfu/flexfec/decoder.go @@ -12,11 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package flexfec implements a FlexFEC-03 decoder used to recover RTP packets -// lost on the publisher leg before they are forwarded downstream. +// Package flexfec recovers lost upstream RTP packets and generates downstream +// repair packets using FlexFEC-03. // https://datatracker.ietf.org/doc/html/draft-ietf-payload-flexible-fec-scheme-03 // -// The recovery logic is ported from pion/interceptor pkg/flexfec +// The decoder's recovery logic is ported from pion/interceptor pkg/flexfec // (https://github.com/pion/interceptor, MIT License, Copyright The Pion // community), which is itself modeled on libwebrtc's ForwardErrorCorrection // receiver. Deviations from the pion implementation: diff --git a/pkg/sfu/flexfec/encoder.go b/pkg/sfu/flexfec/encoder.go new file mode 100644 index 000000000..60de5c631 --- /dev/null +++ b/pkg/sfu/flexfec/encoder.go @@ -0,0 +1,161 @@ +// 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 flexfec + +import ( + "math/rand/v2" + "sync" + "time" + + pionflexfec "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtp" + + "github.com/livekit/protocol/utils/mono" +) + +const ( + // DefaultProtectionPercent is the repair-to-media packet ratio when unspecified. + DefaultProtectionPercent uint32 = 0 + MaxProtectionPercent uint32 = 100 + // MediaPacketsPerGroup bounds both retained packet memory and encoding work. + MediaPacketsPerGroup = 5 + maxEncoderGroupAge = 200 * time.Millisecond + // Reserve the FEC header and up to 16 bytes of outer RTP extensions (AST/TWCC). + // Groups of at most 15 packets use only the first FlexFEC packet mask. + maxEncoderMediaPacketSize = maxMediaPacketSize - pionflexfec.BaseFec03HeaderSize - 16 +) + +// Encoder batches final, outgoing video packets for Pion's FlexFEC-03 encoder. +// It owns reusable wire storage: the caller may recycle headers, extensions and +// payloads as soon as Encode returns. It does not delay media or start workers. +// Methods are safe to call from concurrent pass-through pacer writes. +type Encoder struct { + mu sync.Mutex + encoder *pionflexfec.FlexEncoder03 + packets [MediaPacketsPerGroup]rtp.Packet + storage [MediaPacketsPerGroup][maxEncoderMediaPacketSize]byte + count int + startedAt int64 + sequenceNumber uint16 + closed bool + onSent func(packets int, bytes int) + protectionPercent uint32 + repairCredit uint32 +} + +func NewEncoder(payloadType uint8, ssrc uint32, onSent func(packets int, bytes int)) *Encoder { + return &Encoder{ + encoder: pionflexfec.NewFlexEncoder03(payloadType, ssrc), + sequenceNumber: uint16(rand.Uint32()), + onSent: onSent, + protectionPercent: DefaultProtectionPercent, + } +} + +// Encode returns independently owned repair packets for complete groups only. +// Gaps (including skipped padding), codec switches and stale groups start a new +// group, since Pion requires consecutive sequence numbers. RTP wrap is valid. +// Call after the media write, including any interceptor header modifications. +func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { + e.mu.Lock() + defer e.mu.Unlock() + if e.closed || e.protectionPercent == 0 { + return nil + } + + now := mono.UnixNano() + size := header.MarshalSize() + len(payload) + if header.Padding || len(payload) == 0 || size > maxEncoderMediaPacketSize { + e.count = 0 + return nil + } + if e.count != 0 { + previous := &e.packets[e.count-1] + if header.SequenceNumber != previous.SequenceNumber+1 || + header.SSRC != previous.SSRC || header.PayloadType != previous.PayloadType || + now-e.startedAt > int64(maxEncoderGroupAge) { + e.count = 0 + } + } + if e.count == 0 { + e.startedAt = now + } + + raw := e.storage[e.count][:size] + n, err := header.MarshalTo(raw) + if err != nil { + e.count = 0 + return nil + } + copy(raw[n:], payload) + if err = e.packets[e.count].Unmarshal(raw); err != nil { + e.count = 0 + return nil + } + e.count++ + if e.count < MediaPacketsPerGroup { + return nil + } + + e.count = 0 + // Carry fractional packets between groups rather than rounding each group + // up (which would turn e.g. 30% into 40%). Storage and latency stay bounded + // at five media packets, independent of the configured percentage. + e.repairCredit += e.protectionPercent * MediaPacketsPerGroup + numRepair := e.repairCredit / 100 + e.repairCredit %= 100 + if numRepair == 0 { + return nil + } + repair := e.encoder.EncodeFec(e.packets[:], numRepair) + for i := range repair { + // Pion's encoder uses a constant timestamp and initial sequence number. + // Use this stream's media clock and a randomized, continuous repair sequence. + repair[i].Timestamp = header.Timestamp + repair[i].SequenceNumber = e.sequenceNumber + e.sequenceNumber++ + } + return repair +} + +// SetProtectionPercent updates the packet ratio without changing the repair +// SSRC or sequence number. A new value discards partial groups and old credit. +// Values above 100 are capped; zero disables encoding before any packet copies. +func (e *Encoder) SetProtectionPercent(percent uint32) { + percent = min(percent, MaxProtectionPercent) + e.mu.Lock() + if e.protectionPercent != percent { + e.protectionPercent = percent + e.count = 0 + e.repairCredit = 0 + } + e.mu.Unlock() +} + +// RecordSent reports successful repair writes, with RTP payload bytes like the +// upstream counters. Callbacks run outside the encoder lock. +func (e *Encoder) RecordSent(packets int, bytes int) { + if e.onSent != nil && packets != 0 { + e.onSent(packets, bytes) + } +} + +// Close prevents queued packets from generating repairs after unbind/close. +func (e *Encoder) Close() { + e.mu.Lock() + e.closed = true + e.count = 0 + e.mu.Unlock() +} diff --git a/pkg/sfu/flexfec/encoder_test.go b/pkg/sfu/flexfec/encoder_test.go new file mode 100644 index 000000000..20188cefb --- /dev/null +++ b/pkg/sfu/flexfec/encoder_test.go @@ -0,0 +1,294 @@ +// 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 flexfec + +import ( + "fmt" + "sync" + "testing" + + "github.com/livekit/protocol/logger" + "github.com/pion/rtp" + "github.com/stretchr/testify/require" +) + +func TestEncoderDefaultsToNoProtection(t *testing.T) { + encoder := NewEncoder(testFECPT, testFECSSRC, nil) + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} + for range 100 { + p.SequenceNumber++ + require.Empty(t, encoder.Encode(&p, []byte{1, 2, 3})) + } + require.Zero(t, encoder.count, "disabled protection must not retain media") +} + +func TestEncoderRecoveryWithReusedMemory(t *testing.T) { + for _, base := range []uint16{100, 65533} { + media := makeMediaPackets(t, base, 2*MediaPacketsPerGroup) + encoder := NewEncoder(testFECPT, testFECSSRC, nil) + encoder.SetProtectionPercent(20) + encoder.sequenceNumber = 65535 + decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + var repair []rtp.Packet + for i := range media { + media[i].CSRC = []uint32{123, 456} + require.NoError(t, media[i].SetExtension(3, []byte{byte(i), 2, 3})) + require.NoError(t, media[i].SetExtension(5, []byte{7, byte(i)})) + packet := media[i].Clone() + repair = append(repair, encoder.Encode(&packet.Header, packet.Payload)...) + // The caller owns and immediately recycles every part of the input. + clear(packet.Payload) + clear(packet.CSRC) + clear(packet.GetExtension(3)) + packet.Header = rtp.Header{} + if i != 2 && i != 7 { + decoder.DecodeFEC(&media[i]) + } + } + require.Len(t, repair, 2) + require.Equal(t, uint16(65535), repair[0].SequenceNumber) + require.Zero(t, repair[1].SequenceNumber) + for i := range repair { + require.Equal(t, media[(i+1)*MediaPacketsPerGroup-1].Timestamp, repair[i].Timestamp) + recovered := decoder.DecodeFEC(&repair[i]) + require.Len(t, recovered, 1) + expected, err := media[i*MediaPacketsPerGroup+2].Marshal() + require.NoError(t, err) + actual, err := recovered[0].Marshal() + require.NoError(t, err) + require.Equal(t, expected, actual) + } + } +} + +func TestEncoderGroupBoundaries(t *testing.T) { + for _, name := range []string{"gap", "duplicate", "out of order", "codec", "ssrc", "stale", "oversized", "padding", "empty"} { + t.Run(name, func(t *testing.T) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(20) + media := makeMediaPackets(t, 100, 10) + for i := 0; i < 4; i++ { + require.Empty(t, e.Encode(&media[i].Header, media[i].Payload)) + } + p := &media[4] + switch name { + case "gap": + p.SequenceNumber++ + case "duplicate": + p.SequenceNumber-- + case "out of order": + p.SequenceNumber -= 2 + case "codec": + p.PayloadType++ + case "ssrc": + p.SSRC++ + case "stale": + e.startedAt -= int64(maxEncoderGroupAge) + case "oversized": + p.Payload = make([]byte, maxEncoderMediaPacketSize) + case "padding": + p.Padding = true + case "empty": + p.Payload = nil + } + require.Empty(t, e.Encode(&p.Header, p.Payload), "must not protect a discontinuous group") + fresh := makeMediaPackets(t, 200, MediaPacketsPerGroup) + for i := range fresh { + out := e.Encode(&fresh[i].Header, fresh[i].Payload) + if i == MediaPacketsPerGroup-1 { + require.Len(t, out, 1, "resume protection after the discontinuity") + } else { + require.Empty(t, out) + } + } + }) + } +} + +func TestEncoderSizeLimitAndClose(t *testing.T) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(20) + p := &rtp.Packet{Header: rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}, Payload: make([]byte, maxEncoderMediaPacketSize-12)} + for i := range MediaPacketsPerGroup { + p.SequenceNumber++ + out := e.Encode(&p.Header, p.Payload) + if i == MediaPacketsPerGroup-1 { + require.Len(t, out, 1) + require.NoError(t, out[0].SetExtension(3, []byte{1, 2, 3})) + require.NoError(t, out[0].SetExtension(5, []byte{1, 2})) + require.LessOrEqual(t, out[0].MarshalSize(), maxMediaPacketSize) + } + } + e.Close() + for range 2 * MediaPacketsPerGroup { + p.SequenceNumber++ + require.Empty(t, e.Encode(&p.Header, p.Payload)) + } +} + +func TestEncoderConcurrentClose(t *testing.T) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(20) + var wg sync.WaitGroup + for n := range 4 { + wg.Go(func() { + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, SequenceNumber: uint16(n * 1000)} + for range 100 { + e.Encode(&p, []byte{1, 2, 3}) + p.SequenceNumber++ + } + }) + } + e.Close() + wg.Wait() +} + +func TestEncoderAccumulationReusesStorage(t *testing.T) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(20) + p := rtp.Header{Version: 2, SSRC: testMediaSSRC} + require.NoError(t, p.SetExtension(3, []byte{1, 2, 3})) + payload := make([]byte, 1200) + allocs := testing.AllocsPerRun(100, func() { + e.count = 0 + for range MediaPacketsPerGroup - 1 { + p.SequenceNumber++ + e.Encode(&p, payload) + } + }) + require.Zero(t, allocs) +} + +func BenchmarkEncoder(b *testing.B) { + for _, percent := range []uint32{0, 15, 25, 35} { + b.Run(fmt.Sprintf("percent_%d", percent), func(b *testing.B) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(percent) + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} + _ = p.SetExtension(3, []byte{1, 2, 3}) + payload := make([]byte, 1200) + // Warm the reusable packet slots and Pion's coverage cache. + for range 2 * MediaPacketsPerGroup { + p.SequenceNumber++ + e.Encode(&p, payload) + } + b.ReportAllocs() + b.SetBytes(int64(MediaPacketsPerGroup * len(payload))) + b.ResetTimer() + for b.Loop() { + for range MediaPacketsPerGroup { + p.SequenceNumber++ + e.Encode(&p, payload) + } + } + }) + } +} + +func TestEncoderProtectionPercent(t *testing.T) { + for percent := uint32(0); percent <= 101; percent++ { + t.Run(fmt.Sprint(percent), func(t *testing.T) { + encoder := NewEncoder(testFECPT, testFECSSRC, nil) + encoder.SetProtectionPercent(percent) + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} + var count int + var previous uint16 + for range 100 { + p.SequenceNumber++ + repairs := encoder.Encode(&p, []byte{1, 2, 3}) + require.LessOrEqual(t, len(repairs), MediaPacketsPerGroup) + for _, repair := range repairs { + if count > 0 { + require.Equal(t, previous+1, repair.SequenceNumber) + } + count++ + previous = repair.SequenceNumber + } + } + require.EqualValues(t, min(percent, MaxProtectionPercent), count) + }) + } +} + +func TestEncoderProtectionPercentChanges(t *testing.T) { + encoder := NewEncoder(testFECPT, testFECSSRC, nil) + encoder.SetProtectionPercent(20) + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} + send := func(count int) []rtp.Packet { + var result []rtp.Packet + for range count { + p.SequenceNumber++ + result = append(result, encoder.Encode(&p, []byte{1, 2, 3})...) + } + return result + } + first := send(MediaPacketsPerGroup) + require.Len(t, first, 1) + require.Empty(t, send(4)) + encoder.SetProtectionPercent(40) + require.Empty(t, send(4), "discard the old partial group") + repairs := send(1) + require.Len(t, repairs, 2) + require.Equal(t, first[0].SequenceNumber+1, repairs[0].SequenceNumber) + encoder.SetProtectionPercent(0) + require.Empty(t, send(100)) + require.Zero(t, encoder.count) + encoder.SetProtectionPercent(10) + require.Empty(t, send(5)) + encoder.SetProtectionPercent(10) // unchanged settings must retain fractional credit + last := send(5) + require.Len(t, last, 1) + require.Equal(t, repairs[1].SequenceNumber+1, last[0].SequenceNumber) +} + +func TestEncoderMultipleRepairRecovery(t *testing.T) { + encoder := NewEncoder(testFECPT, testFECSSRC, nil) + encoder.SetProtectionPercent(60) + media := makeMediaPackets(t, 100, MediaPacketsPerGroup) + decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + var repair []rtp.Packet + for i := range media { + repair = append(repair, encoder.Encode(&media[i].Header, media[i].Payload)...) + if i >= 3 { + decoder.DecodeFEC(&media[i]) + } + } + require.Len(t, repair, 3) + for i := range repair { + recovered := decoder.DecodeFEC(&repair[i]) + require.Len(t, recovered, 1) + requirePacketEqual(t, &media[i], recovered[0]) + } +} + +func TestEncoderConcurrentProtectionChanges(t *testing.T) { + encoder := NewEncoder(testFECPT, testFECSSRC, nil) + encoder.SetProtectionPercent(20) + var wg sync.WaitGroup + wg.Go(func() { + for i := range 1000 { + encoder.SetProtectionPercent(uint32(i % 101)) + } + }) + wg.Go(func() { + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} + for range 1000 { + p.SequenceNumber++ + encoder.Encode(&p, []byte{1, 2, 3}) + } + }) + wg.Wait() + encoder.Close() +} diff --git a/pkg/sfu/pacer/base.go b/pkg/sfu/pacer/base.go index 96e3be250..f1ea1aae2 100644 --- a/pkg/sfu/pacer/base.go +++ b/pkg/sfu/pacer/base.go @@ -85,6 +85,39 @@ func (b *Base) SendPacket(p *Packet) (int, error) { return 0, err } + if p.FEC != nil && written > 0 && !p.IsRTX && !p.IsProbe { + repair := p.FEC.Encode(p.Header, p.Payload) + sent, payloadBytes := 0, 0 + for i := range repair { + packet := &repair[i] + fec := PacketFactory.Get().(*Packet) + *fec = Packet{ + Header: &packet.Header, + Payload: packet.Payload, + AbsSendTimeExtID: p.AbsSendTimeExtID, + TransportWideExtID: p.TransportWideExtID, + WriteStream: p.WriteStream, + ProbeClusterId: p.ProbeClusterId, + } + // Include extension space in BWE/probe accounting before patching. + if fec.AbsSendTimeExtID != 0 { + _ = fec.Header.SetExtension(fec.AbsSendTimeExtID, []byte{0, 0, 0}) + } + if fec.TransportWideExtID != 0 { + _ = fec.Header.SetExtension(fec.TransportWideExtID, []byte{0, 0}) + } + fec.HeaderSize = fec.Header.MarshalSize() + // Charge repair bytes to this send's pacer budget. Repairs immediately + // follow their media group even when there is a long media queue. + n, fecErr := b.SendPacket(fec) + written += n + if fecErr == nil && n > 0 { + sent++ + payloadBytes += len(packet.Payload) + } + } + p.FEC.RecordSent(sent, payloadBytes) + } return written, nil } diff --git a/pkg/sfu/pacer/base_fec_test.go b/pkg/sfu/pacer/base_fec_test.go new file mode 100644 index 000000000..17252ec7f --- /dev/null +++ b/pkg/sfu/pacer/base_fec_test.go @@ -0,0 +1,145 @@ +// 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 pacer + +import ( + "encoding/binary" + "errors" + "io" + "testing" + + "github.com/livekit/livekit-server/pkg/sfu/bwe" + "github.com/livekit/livekit-server/pkg/sfu/ccutils" + "github.com/livekit/livekit-server/pkg/sfu/flexfec" + "github.com/livekit/protocol/logger" + "github.com/pion/rtp" + "github.com/stretchr/testify/require" +) + +type fecTestBWE struct { + bwe.BWE + sizes []int +} + +func (b *fecTestBWE) RecordPacketSendAndGetSequenceNumber(_ int64, size int, _ bool, _ ccutils.ProbeClusterId, _ bool) uint16 { + b.sizes = append(b.sizes, size) + return uint16(len(b.sizes)) +} + +type fecTestWriter struct { + packets []*rtp.Packet + failMedia bool + failRepair bool + noWriter bool + noRepairWriter bool +} + +func (w *fecTestWriter) WriteRTP(h *rtp.Header, payload []byte) (int, error) { + if w.noWriter || (w.noRepairWriter && h.SSRC == 456) { + // Pion returns (0, nil) until its interceptor writer is installed. + return 0, nil + } + if (w.failMedia && h.SSRC == 123) || (w.failRepair && h.SSRC == 456) { + return 0, io.ErrClosedPipe + } + // Model an interceptor changing a header after the pacer has patched it. + if h.SSRC == 123 { + _ = h.SetExtension(7, []byte{byte(h.SequenceNumber), 9}) + } + p := (&rtp.Packet{Header: *h, Payload: payload}).Clone() + w.packets = append(w.packets, p) + return p.MarshalSize(), nil +} +func (w *fecTestWriter) Write([]byte) (int, error) { panic("unexpected raw write") } + +func TestPacerFECFinalHeadersAndAccounting(t *testing.T) { + bw := &fecTestBWE{} + b := NewBase(logger.GetLogger(), bw) + w := &fecTestWriter{} + sent, payloadBytes := 0, 0 + encoder := flexfec.NewEncoder(115, 456, func(n int, bytes int) { sent += n; payloadBytes += bytes }) + encoder.SetProtectionPercent(20) + total := 0 + for i := range flexfec.MediaPacketsPerGroup { + p := PacketFactory.Get().(*Packet) + h := &rtp.Header{Version: 2, SSRC: 123, PayloadType: 96, SequenceNumber: uint16(i), Timestamp: uint32(i * 3000)} + _ = h.SetExtension(3, []byte{0, 0, 0}) + _ = h.SetExtension(5, []byte{0, 0}) + _ = h.SetExtension(7, []byte{0, 0}) + *p = Packet{Header: h, HeaderSize: h.MarshalSize(), Payload: []byte{byte(i), 4, 5}, WriteStream: w, FEC: encoder, AbsSendTimeExtID: 3, TransportWideExtID: 5} + n, err := b.SendPacket(p) + require.NoError(t, err) + total += n + } + require.Len(t, w.packets, 6) + require.Len(t, bw.sizes, 6) + require.Equal(t, 1, sent) + require.Equal(t, len(w.packets[5].Payload), payloadBytes) + wireBytes := 0 + media := map[uint16][]byte{} + for i, p := range w.packets { + wireBytes += p.MarshalSize() + require.Equal(t, p.MarshalSize(), bw.sizes[i], "BWE must account for every media and repair byte") + require.EqualValues(t, i+1, binary.BigEndian.Uint16(p.GetExtension(5))) + if i < 5 && i != 2 { + media[p.SequenceNumber], _ = p.Marshal() + } + } + require.Equal(t, wireBytes, total, "repair bytes must consume the pacer budget") + decoder := flexfec.NewDecoder(456, 123, func(sn uint16, dst []byte) (int, error) { + raw, ok := media[sn] + if !ok { + return 0, errors.New("missing") + } + return copy(dst, raw), nil + }, logger.GetLogger()) + recovered := decoder.DecodeFEC(w.packets[5]) + require.Len(t, recovered, 1) + expected, _ := w.packets[2].Marshal() + actual, _ := recovered[0].Marshal() + require.Equal(t, expected, actual, "FEC must cover post-interceptor headers") +} + +func TestPacerFECSkipsRTXProbesAndFailedWrites(t *testing.T) { + for _, mode := range []string{"rtx", "probe", "media failure", "repair failure", "writer not ready", "repair writer not ready"} { + t.Run(mode, func(t *testing.T) { + b := NewBase(logger.GetLogger(), nil) + w := &fecTestWriter{failMedia: mode == "media failure", failRepair: mode == "repair failure", noWriter: mode == "writer not ready", noRepairWriter: mode == "repair writer not ready"} + sent := 0 + e := flexfec.NewEncoder(115, 456, func(n int, _ int) { sent += n }) + e.SetProtectionPercent(20) + for i := range 10 { + p := PacketFactory.Get().(*Packet) + *p = Packet{Header: &rtp.Header{Version: 2, SSRC: 123, SequenceNumber: uint16(i)}, HeaderSize: 12, Payload: []byte{1}, FEC: e, WriteStream: w, IsRTX: mode == "rtx", IsProbe: mode == "probe"} + n, err := b.SendPacket(p) + if w.failMedia { + require.ErrorIs(t, err, io.ErrClosedPipe) + require.Zero(t, n) + } else { + require.NoError(t, err) + if w.noWriter { + require.Zero(t, n) + } else { + require.Positive(t, n) + } + } + } + require.Zero(t, sent) + for _, p := range w.packets { + require.EqualValues(t, 123, p.SSRC) + } + }) + } +} diff --git a/pkg/sfu/pacer/pacer.go b/pkg/sfu/pacer/pacer.go index f23212519..bf02ce988 100644 --- a/pkg/sfu/pacer/pacer.go +++ b/pkg/sfu/pacer/pacer.go @@ -19,6 +19,7 @@ import ( "time" "github.com/livekit/livekit-server/pkg/sfu/ccutils" + "github.com/livekit/livekit-server/pkg/sfu/flexfec" "github.com/pion/rtp" "github.com/pion/webrtc/v4" ) @@ -42,6 +43,8 @@ const ( ) type Packet struct { + // FEC is set only on original video media, never padding, RTX or probes. + FEC *flexfec.Encoder Header *rtp.Header HeaderPool *sync.Pool HeaderSize int diff --git a/pkg/telemetry/prometheus/packets.go b/pkg/telemetry/prometheus/packets.go index 406b6bdbc..8e3096460 100644 --- a/pkg/telemetry/prometheus/packets.go +++ b/pkg/telemetry/prometheus/packets.go @@ -243,6 +243,12 @@ func RecordFECUpstream(received int, recovered int, discarded int, bytesReceived } } +// RecordFECDownstream tracks successfully sent repair packets and RTP payload bytes. +func RecordFECDownstream(packets int, bytes int) { + promFECPacketTotal.WithLabelValues(string(Outgoing), "sent").Add(float64(packets)) + promFECPacketBytes.WithLabelValues(string(Outgoing), "sent").Add(float64(bytes)) +} + func IncrementPackets(country string, direction Direction, count uint64, retransmit bool) { var transmission TransmissionType if retransmit {