From 195adeb38c8d6c5a65424b23150ecce77914cc05 Mon Sep 17 00:00:00 2001 From: David Chen Date: Mon, 15 Jun 2026 13:41:04 -0700 Subject: [PATCH] Add FlexFEC-03 support to the SFU: upstream recovery and downstream generation Implements FlexFEC-03 (RFC 8627) end to end in the SFU, in both directions: - Upstream: decode FlexFEC repair packets from publishers and recover lost media in the receive buffer before forwarding (pkg/sfu/flexfec/decoder.go, pkg/sfu/buffer). Recovered packets are surfaced via Prometheus counters. - Downstream: generate FlexFEC for subscribers off the downtrack, paced alongside media (pkg/sfu/downtrack_fec.go, pkg/sfu/pacer). - Negotiation: advertise and match flexfec-03 in the media engine and transport SDP, gated by new config knobs (pkg/rtc, pkg/config, config-sample.yaml). - Telemetry: livekit_fec_* metrics for sent/received/recovered packets (pkg/telemetry/prometheus/packets.go). Tests: unit coverage for the decoder, buffer recovery, downtrack generation, and transport negotiation, plus an end-to-end integration test and the test-client support it needs (test/flexfec_test.go, test/client). --- config-sample.yaml | 14 + pkg/config/config.go | 52 +++ pkg/rtc/config.go | 45 ++- pkg/rtc/mediaengine.go | 64 ++++ pkg/rtc/mediatrack.go | 5 + pkg/rtc/subscribedtrack.go | 5 + pkg/rtc/transport.go | 63 ++++ pkg/rtc/transport_fec_test.go | 125 +++++++ pkg/service/egress.go | 43 ++- pkg/sfu/buffer/buffer.go | 168 ++++++++++ pkg/sfu/buffer/buffer_base.go | 9 + pkg/sfu/buffer/buffer_fec_test.go | 324 ++++++++++++++++++ pkg/sfu/buffer/factory.go | 30 ++ pkg/sfu/downtrack.go | 110 ++++++ pkg/sfu/downtrack_fec.go | 134 ++++++++ pkg/sfu/downtrack_fec_test.go | 206 ++++++++++++ pkg/sfu/flexfec/decoder.go | 504 ++++++++++++++++++++++++++++ pkg/sfu/flexfec/decoder_test.go | 355 ++++++++++++++++++++ pkg/sfu/pacer/base.go | 7 +- pkg/sfu/pacer/pacer.go | 5 + pkg/sfu/utils/helpers.go | 13 + pkg/telemetry/prometheus/packets.go | 46 +++ test/client/client.go | 19 ++ test/flexfec_test.go | 178 ++++++++++ 24 files changed, 2507 insertions(+), 17 deletions(-) create mode 100644 pkg/rtc/transport_fec_test.go create mode 100644 pkg/sfu/buffer/buffer_fec_test.go create mode 100644 pkg/sfu/downtrack_fec.go create mode 100644 pkg/sfu/downtrack_fec_test.go create mode 100644 pkg/sfu/flexfec/decoder.go create mode 100644 pkg/sfu/flexfec/decoder_test.go create mode 100644 test/flexfec_test.go diff --git a/config-sample.yaml b/config-sample.yaml index e4fd969c2..e428fad7b 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -112,6 +112,20 @@ rtc: # # in the unlikely event of highly congested networks, SFU may choose to pause some tracks # # in order to allow others to stream smoothly. You can disable this behavior here # allow_pause: true + # # FlexFEC-03 forward error correction, both legs disabled by default. + # # Upstream accepts FEC from publishers (e.g. the Rust SDK with FlexFEC + # # enabled) and uses it to repair lost packets before forwarding. + # # Downstream generates FEC for forwarded video so subscribers can repair + # # losses without waiting for retransmissions. Overhead per protected + # # downstream track is roughly num_fec_packets/num_media_packets. + # flexfec: + # upstream_enabled: false + # downstream_enabled: false + # # payload type offered to subscribers, must not collide with codec or RTX payload types + # payload_type: 115 + # # downstream protection window: FEC packets generated per media packets + # num_media_packets: 10 + # num_fec_packets: 2 # # 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/pkg/config/config.go b/pkg/config/config.go index 692c0af38..059420cf8 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -114,6 +114,8 @@ type RTCConfig struct { CongestionControl CongestionControlConfig `yaml:"congestion_control,omitempty"` + FlexFEC FlexFECConfig `yaml:"flexfec,omitempty"` + // allow TCP and TURN/TLS fallback AllowTCPFallback *bool `yaml:"allow_tcp_fallback,omitempty"` @@ -185,6 +187,55 @@ type CongestionControlConfig struct { SendSideBWE sendsidebwe.SendSideBWEConfig `yaml:"send_side_bwe,omitempty"` } +// FlexFECConfig controls FlexFEC-03 forward error correction. +// The upstream (publisher -> SFU) and downstream (SFU -> subscriber) legs are +// independent: upstream accepts and decodes FEC sent by publishers to repair +// lost packets before forwarding, downstream generates fresh FEC for +// forwarded video. +type FlexFECConfig struct { + // negotiate flexfec-03 with publishers and use it to recover lost upstream packets + UpstreamEnabled bool `yaml:"upstream_enabled,omitempty"` + // offer flexfec-03 to subscribers and generate FEC for forwarded video + DownstreamEnabled bool `yaml:"downstream_enabled,omitempty"` + // payload type used when offering flexfec-03 to subscribers, must not + // collide with codec payload types or their RTX (pt+1) slots + PayloadType uint8 `yaml:"payload_type,omitempty"` + // downstream: media packets accumulated per FEC batch + NumMediaPackets uint32 `yaml:"num_media_packets,omitempty"` + // downstream: FEC packets generated per batch (overhead ≈ num_fec_packets/num_media_packets) + NumFECPackets uint32 `yaml:"num_fec_packets,omitempty"` +} + +var DefaultFlexFECConfig = FlexFECConfig{ + UpstreamEnabled: false, + DownstreamEnabled: false, + PayloadType: 115, + NumMediaPackets: 10, + NumFECPackets: 2, +} + +// WithDefaults returns a copy with zero values replaced by defaults and +// out-of-range values clamped. +func (c FlexFECConfig) WithDefaults() FlexFECConfig { + if c.PayloadType == 0 { + c.PayloadType = DefaultFlexFECConfig.PayloadType + } + if c.NumMediaPackets == 0 { + c.NumMediaPackets = DefaultFlexFECConfig.NumMediaPackets + } + // flexfec-03 masks cover at most 109 media packets, stay well below + if c.NumMediaPackets > 48 { + c.NumMediaPackets = 48 + } + if c.NumFECPackets == 0 { + c.NumFECPackets = DefaultFlexFECConfig.NumFECPackets + } + if c.NumFECPackets > c.NumMediaPackets { + c.NumFECPackets = c.NumMediaPackets + } + return c +} + type PlayoutDelayConfig struct { Enabled bool `yaml:"enabled,omitempty"` Min int `yaml:"min,omitempty"` @@ -410,6 +461,7 @@ var DefaultConfig = Config{ SendSideBWEPacer: string(pacer.PacerBehaviorNoQueue), SendSideBWE: sendsidebwe.DefaultSendSideBWEConfig, }, + FlexFEC: DefaultFlexFECConfig, }, Audio: sfu.DefaultAudioConfig, Video: VideoConfig{ diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index 2f30060fd..456e6d74e 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -37,6 +37,8 @@ type WebRTCConfig struct { Receiver ReceiverConfig Publisher DirectionConfig Subscriber DirectionConfig + + flexFEC config.FlexFECConfig } type ReceiverConfig struct { @@ -57,6 +59,18 @@ type RTCPFeedbackConfig struct { type DirectionConfig struct { RTPHeaderExtension RTPHeaderExtensionConfig RTCPFeedback RTCPFeedbackConfig + FlexFEC FlexFECDirectionConfig +} + +// FlexFECDirectionConfig enables flexfec-03 for the leg this DirectionConfig +// governs: recovery of publisher FEC on the publisher (upstream) side, +// generation of FEC toward subscribers on the subscriber (downstream) side. +type FlexFECDirectionConfig struct { + Enabled bool + PayloadType uint8 + // downstream generation parameters, unused on the upstream leg + NumMediaPackets uint32 + NumFECPackets uint32 } func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { @@ -80,23 +94,30 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { rtcConf.PacketBufferSizeAudio = rtcConf.PacketBufferSize } + flexFEC := rtcConf.FlexFEC.WithDefaults() + if flexFEC.UpstreamEnabled || flexFEC.DownstreamEnabled { + if err := validateFlexFECPayloadType(flexFEC.PayloadType); err != nil { + return nil, err + } + } return &WebRTCConfig{ WebRTCConfig: *webRTCConfig, Receiver: ReceiverConfig{ PacketBufferSizeVideo: rtcConf.PacketBufferSizeVideo, PacketBufferSizeAudio: rtcConf.PacketBufferSizeAudio, }, - Publisher: getPublisherConfig(false), - Subscriber: getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE), + Publisher: getPublisherConfig(false, flexFEC), + Subscriber: getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE, flexFEC), + flexFEC: flexFEC, }, nil } func (c *WebRTCConfig) UpdatePublisherConfig(consolidated bool) { - c.Publisher = getPublisherConfig(consolidated) + c.Publisher = getPublisherConfig(consolidated, c.flexFEC) } 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) { @@ -104,9 +125,14 @@ func (c *WebRTCConfig) SetBufferFactory(factory *buffer.Factory) { c.SettingEngine.BufferFactory = factory.GetOrNew } -func getPublisherConfig(consolidated bool) DirectionConfig { +func getPublisherConfig(consolidated bool, flexFEC config.FlexFECConfig) DirectionConfig { + publisherFlexFEC := FlexFECDirectionConfig{ + Enabled: flexFEC.UpstreamEnabled, + PayloadType: flexFEC.PayloadType, + } if consolidated { return DirectionConfig{ + FlexFEC: publisherFlexFEC, RTPHeaderExtension: RTPHeaderExtensionConfig{ Audio: []string{ sdp.SDESMidURI, @@ -141,6 +167,7 @@ func getPublisherConfig(consolidated bool) DirectionConfig { } return DirectionConfig{ + FlexFEC: publisherFlexFEC, RTPHeaderExtension: RTPHeaderExtensionConfig{ Audio: []string{ sdp.SDESMidURI, @@ -172,8 +199,14 @@ func getPublisherConfig(consolidated bool) DirectionConfig { } } -func getSubscriberConfig(enableTWCC bool) DirectionConfig { +func getSubscriberConfig(enableTWCC bool, flexFEC config.FlexFECConfig) DirectionConfig { subscriberConfig := DirectionConfig{ + FlexFEC: FlexFECDirectionConfig{ + Enabled: flexFEC.DownstreamEnabled, + PayloadType: flexFEC.PayloadType, + NumMediaPackets: flexFEC.NumMediaPackets, + NumFECPackets: flexFEC.NumFECPackets, + }, RTPHeaderExtension: RTPHeaderExtensionConfig{ Video: []string{ dd.ExtensionURI, diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index cfc53eaaf..34b5f7b67 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -25,6 +25,51 @@ import ( "github.com/livekit/protocol/livekit" ) +// flexFECCodecParameters returns the flexfec-03 codec registered/offered when +// FlexFEC is enabled for a direction. Mirrors pion's ConfigureFlexFEC03 codec +// minus its generator interceptor (the SFU runs its own encode/decode paths). +func flexFECCodecParameters(payloadType uint8) webrtc.RTPCodecParameters { + return webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeFlexFEC03, + ClockRate: 90000, + SDPFmtpLine: "repair-window=10000000", + RTCPFeedback: []webrtc.RTCPFeedback{ + {Type: webrtc.TypeRTCPFBTransportCC}, + }, + }, + PayloadType: webrtc.PayloadType(payloadType), + } +} + +func isFlexFEC03MimeType(mimeType string) bool { + return strings.EqualFold(mimeType, webrtc.MimeTypeFlexFEC03) +} + +// validateFlexFECPayloadType ensures the configured flexfec payload type does +// not collide with any known codec payload type or its RTX (pt+1) slot. +func validateFlexFECPayloadType(payloadType uint8) error { + pt := webrtc.PayloadType(payloadType) + for _, codec := range protoCodecs.VideoCodecsParameters { + if pt == codec.PayloadType || pt == codec.PayloadType+1 { + return fmt.Errorf("flexfec payload type %d collides with %s (pt %d / rtx pt %d)", + payloadType, codec.MimeType, codec.PayloadType, codec.PayloadType+1) + } + } + for _, codec := range []webrtc.RTPCodecParameters{ + protoCodecs.OpusCodecParameters, + protoCodecs.RedCodecParameters, + protoCodecs.PCMUCodecParameters, + protoCodecs.PCMACodecParameters, + } { + if pt == codec.PayloadType { + return fmt.Errorf("flexfec payload type %d collides with %s (pt %d)", + payloadType, codec.MimeType, codec.PayloadType) + } + } + return nil +} + func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedback RTCPFeedbackConfig, filterOutH264HighProfile bool) error { // audio codecs if IsCodecEnabled(codecs, protoCodecs.OpusCodecParameters.RTPCodecCapability) { @@ -108,6 +153,15 @@ func createMediaEngine(codecs []*livekit.Codec, config DirectionConfig, filterOu return nil, err } + if config.FlexFEC.Enabled { + // registering a flexfec codec makes pion allocate FEC SSRCs for video + // senders and emit a=ssrc-group:FEC-FR in offers, and lets answers + // accept flexfec offered by publishers + if err := me.RegisterCodec(flexFECCodecParameters(config.FlexFEC.PayloadType), webrtc.RTPCodecTypeVideo); err != nil { + return nil, err + } + } + if err := registerHeaderExtensions(me, config.RTPHeaderExtension); err != nil { return nil, err } @@ -171,6 +225,7 @@ func filterCodecs( enabledCodecs []*livekit.Codec, rtcpFeedbackConfig RTCPFeedbackConfig, filterOutH264HighProfile bool, + keepFlexFEC bool, ) []webrtc.RTPCodecParameters { filteredCodecs := make([]webrtc.RTPCodecParameters, 0, len(codecs)) for _, c := range codecs { @@ -178,6 +233,15 @@ func filterCodecs( continue } + // flexfec-03 is not part of the enabled codec lists, retain it when + // the transport direction has FlexFEC enabled + if isFlexFEC03MimeType(c.RTPCodecCapability.MimeType) { + if keepFlexFEC { + filteredCodecs = append(filteredCodecs, c) + } + continue + } + for _, enabledCodec := range enabledCodecs { if mime.NormalizeMimeType(enabledCodec.Mime) == mime.NormalizeMimeType(c.RTPCodecCapability.MimeType) { if !mime.IsMimeTypeStringEqual(c.RTPCodecCapability.MimeType, mime.MimeTypeRTX.String()) { diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index eefa1c0e3..ed1e50f08 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -37,6 +37,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "github.com/livekit/livekit-server/pkg/sfu/interceptor" "github.com/livekit/livekit-server/pkg/telemetry" + "github.com/livekit/livekit-server/pkg/telemetry/prometheus" util "github.com/livekit/mediatransportutil" ) @@ -589,6 +590,10 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe buff.OnNotifyRTX(t.MediaTrackReceiver.setLayerRtxInfo) + buff.OnFECRecovery(func(recovered int, received int, discarded int, bytesReceived int) { + prometheus.RecordFECUpstream(received, recovered, discarded, uint64(bytesReceived)) + }) + // if subscriber request fps before fps calculated, update them after fps updated. buff.OnFpsChanged(func() { t.MediaTrackSubscriptions.UpdateVideoLayers() diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 3a8a06148..fbc9e4457 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -155,6 +155,11 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error) DisableSenderReportPassThrough: params.Subscriber.GetDisableSenderReportPassThrough(), SupportsCodecChange: params.Subscriber.SupportsCodecChange(), Listener: s, + FlexFEC: sfu.FlexFECParams{ + Enabled: params.SubscriberConfig.FlexFEC.Enabled, + NumMediaPackets: params.SubscriberConfig.FlexFEC.NumMediaPackets, + NumFECPackets: params.SubscriberConfig.FlexFEC.NumFECPackets, + }, }) if err != nil { return nil, err diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 0848ae137..e64c7e7a2 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -1013,6 +1013,7 @@ func (t *PCTransport) queueOrConfigureSender( !t.params.IsOfferer, enableAudioStereo, enableAudioNACK, + t.params.DirectionConfig.FlexFEC.Enabled, } if !t.params.IsOfferer { t.sendersPendingConfigMu.Lock() @@ -1661,6 +1662,15 @@ func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remo t.params.Config.BufferFactory.SetRTXPair(repair, base, "") } } + + if t.params.DirectionConfig.FlexFEC.Enabled { + if fecFlows := fecPairsFromSDP(parsed, t.params.Logger); len(fecFlows) > 0 { + t.params.Logger.Debugw("fec pairs found from sdp", "ssrcs", fecFlows) + for fec, base := range fecFlows { + t.params.Config.BufferFactory.SetFECPair(fec, base) + } + } + } return nil } @@ -2899,6 +2909,15 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, o } } + if t.params.DirectionConfig.FlexFEC.Enabled { + if fecFlows := fecPairsFromSDP(parsed, t.params.Logger); len(fecFlows) > 0 { + t.params.Logger.Debugw("fec pairs found from sdp", "ssrcs", fecFlows) + for fec, base := range fecFlows { + t.params.Config.BufferFactory.SetFECPair(fec, base) + } + } + } + if t.currentOfferIceCredential == "" || offerRestartICE { t.currentOfferIceCredential = iceCredential } @@ -3072,6 +3091,7 @@ type configureSenderParams struct { filterOutH264HighProfile bool enableAudioStereo bool enableAudioNACK bool + keepFlexFEC bool } func configureSender(params configureSenderParams) { @@ -3080,6 +3100,7 @@ func configureSender(params configureSenderParams) { params.enabledCodecs, params.rtcpFeedbackConfig, params.filterOutH264HighProfile, + params.keepFlexFEC, ) if params.transceiver.Kind() == webrtc.RTPCodecTypeAudio { @@ -3132,6 +3153,7 @@ func configureSenderCodecs( enabledCodecs []*livekit.Codec, rtcpFeedbackConfig RTCPFeedbackConfig, filterOutH264HighProfile bool, + keepFlexFEC bool, ) { if len(enabledCodecs) == 0 { return @@ -3147,6 +3169,7 @@ func configureSenderCodecs( enabledCodecs, rtcpFeedbackConfig, filterOutH264HighProfile, + keepFlexFEC, ) tr.SetCodecPreferences(filteredCodecs) } @@ -3169,6 +3192,7 @@ func (t *PCTransport) restrictReceiverCodecsToPublishList() { t.params.EnabledPublishCodecs, t.params.DirectionConfig.RTCPFeedback, false, + t.params.DirectionConfig.FlexFEC.Enabled, ) if len(filtered) == 0 { continue @@ -3220,6 +3244,14 @@ func configureReceiverCodecs( // if the client don't comply with codec order in SDP answer, only keep preferred codecs to force client to use it if compliesWithCodecOrderInSDPAnswer { reorderedCodecs = append(reorderedCodecs, leftCodecs...) + } else { + // flexfec applies to whichever video codec is negotiated, retain + // it even when trimming to the preferred codec + for _, c := range leftCodecs { + if isFlexFEC03MimeType(c.RTPCodecCapability.MimeType) { + reorderedCodecs = append(reorderedCodecs, c) + } + } } } else { reorderedCodecs = append(reorderedCodecs, leftCodecs...) @@ -3270,6 +3302,37 @@ func nonSimulcastRTXRepairsFromSDP(s *sdp.SessionDescription, logger logger.Logg return rtxRepairFlows } +// fecPairsFromSDP extracts FlexFEC repair flows declared via +// `a=ssrc-group:FEC-FR ` (RFC 5956) from the remote +// description, returning a map of fec ssrc -> protected media ssrc. +func fecPairsFromSDP(s *sdp.SessionDescription, logger logger.Logger) map[uint32]uint32 { + fecFlows := map[uint32]uint32{} + for _, media := range s.MediaDescriptions { + for _, attr := range media.Attributes { + if attr.Key != sdp.AttrKeySSRCGroup { + continue + } + split := strings.Split(attr.Value, " ") + if split[0] != sdp.SemanticTokenForwardErrorCorrectionFramework || len(split) != 3 { + continue + } + baseSsrc, err := strconv.ParseUint(split[1], 10, 32) + if err != nil { + logger.Warnw("failed to parse SSRC", err, "ssrc", split[1]) + continue + } + fecSsrc, err := strconv.ParseUint(split[2], 10, 32) + if err != nil { + logger.Warnw("failed to parse SSRC", err, "ssrc", split[2]) + continue + } + fecFlows[uint32(fecSsrc)] = uint32(baseSsrc) + } + } + + return fecFlows +} + // ---------------------- type iceCandidatePairStatsEncoder struct { diff --git a/pkg/rtc/transport_fec_test.go b/pkg/rtc/transport_fec_test.go new file mode 100644 index 000000000..a311c32f3 --- /dev/null +++ b/pkg/rtc/transport_fec_test.go @@ -0,0 +1,125 @@ +// 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/pion/sdp/v3" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +func TestFECPairsFromSDP(t *testing.T) { + offer := `v=0 +o=- 8541913762120318441 2 IN IP4 127.0.0.1 +s=- +t=0 0 +m=video 9 UDP/TLS/RTP/SAVPF 96 97 115 +c=IN IP4 0.0.0.0 +a=mid:0 +a=sendonly +a=rtpmap:96 VP8/90000 +a=rtpmap:97 rtx/90000 +a=fmtp:97 apt=96 +a=rtpmap:115 flexfec-03/90000 +a=fmtp:115 repair-window=10000000 +a=ssrc-group:FID 1111 2222 +a=ssrc-group:FEC-FR 1111 3333 +a=ssrc:1111 cname:test +a=ssrc:2222 cname:test +a=ssrc:3333 cname:test +` + parsed := &sdp.SessionDescription{} + require.NoError(t, parsed.Unmarshal([]byte(offer))) + + fecPairs := fecPairsFromSDP(parsed, logger.GetLogger()) + require.Len(t, fecPairs, 1) + assert.Equal(t, uint32(1111), fecPairs[uint32(3333)]) + + // FID pairs are not picked up as FEC + rtxPairs := nonSimulcastRTXRepairsFromSDP(parsed, logger.GetLogger()) + require.Len(t, rtxPairs, 1) + assert.Equal(t, uint32(1111), rtxPairs[uint32(2222)]) +} + +func TestFECPairsFromSDPNoGroups(t *testing.T) { + offer := `v=0 +o=- 8541913762120318441 2 IN IP4 127.0.0.1 +s=- +t=0 0 +m=video 9 UDP/TLS/RTP/SAVPF 96 +c=IN IP4 0.0.0.0 +a=mid:0 +a=sendonly +a=rtpmap:96 VP8/90000 +a=ssrc:1111 cname:test +` + parsed := &sdp.SessionDescription{} + require.NoError(t, parsed.Unmarshal([]byte(offer))) + assert.Empty(t, fecPairsFromSDP(parsed, logger.GetLogger())) +} + +func TestFlexFECPayloadTypeValidation(t *testing.T) { + assert.NoError(t, validateFlexFECPayloadType(115)) + // VP8 payload type + assert.Error(t, validateFlexFECPayloadType(96)) + // RTX slot of VP8 (pt+1) + assert.Error(t, validateFlexFECPayloadType(97)) + // opus + assert.Error(t, validateFlexFECPayloadType(111)) +} + +func TestMediaEngineRegistersFlexFEC(t *testing.T) { + enabledCodecs := []*livekit.Codec{ + {Mime: "video/VP8"}, + {Mime: "video/rtx"}, + } + + for _, enabled := range []bool{false, true} { + me, err := createMediaEngine(enabledCodecs, DirectionConfig{ + FlexFEC: FlexFECDirectionConfig{ + Enabled: enabled, + PayloadType: 115, + }, + RTCPFeedback: RTCPFeedbackConfig{ + Video: []webrtc.RTCPFeedback{{Type: webrtc.TypeRTCPFBTransportCC}}, + }, + }, false) + require.NoError(t, err) + + // drive codec registration into negotiated form via an SDP round trip + // is heavyweight, instead check via filterCodecs retention behavior + flexfecParams := flexFECCodecParameters(115) + filtered := filterCodecs( + []webrtc.RTPCodecParameters{flexfecParams}, + enabledCodecs, + RTCPFeedbackConfig{}, + false, + enabled, + ) + if enabled { + require.Len(t, filtered, 1) + assert.Equal(t, webrtc.MimeTypeFlexFEC03, filtered[0].MimeType) + } else { + assert.Empty(t, filtered) + } + _ = me + } +} diff --git a/pkg/service/egress.go b/pkg/service/egress.go index 2e37ce4c2..697aecbb4 100644 --- a/pkg/service/egress.go +++ b/pkg/service/egress.go @@ -72,6 +72,33 @@ func NewEgressLauncher(client rpc.EgressClient, io IOClient, store ServiceStore) } } +func (s *EgressService) StartEgress(ctx context.Context, req *livekit.StartEgressRequest) (*livekit.EgressInfo, error) { + sourceType, outputType := egress.GetTypes(&livekit.EgressInfo_Egress{Egress: req}) + fields := []any{ + "room", req.RoomName, + "sourceType", sourceType, + "outputType", outputType, + } + defer func() { + AppendLogFields(ctx, fields...) + }() + + egressID, idFromCtx := EgressID(ctx) + ei, err := s.startEgress(ctx, &rpc.StartEgressRequest{ + EgressId: egressID, + Request: &rpc.StartEgressRequest_Egress{ + Egress: req, + }, + }) + if err != nil { + return nil, err + } + if !idFromCtx { + fields = append(fields, "egressID", ei.EgressId) + } + return ei, err +} + func (s *EgressService) StartRoomCompositeEgress(ctx context.Context, req *livekit.RoomCompositeEgressRequest) (*livekit.EgressInfo, error) { fields := []any{ "room", req.RoomName, @@ -230,6 +257,8 @@ func (s *egressLauncher) StartEgress(ctx context.Context, req *rpc.StartEgressRe roomName = v.TrackComposite.RoomName case *rpc.StartEgressRequest_Track: roomName = v.Track.RoomName + case *rpc.StartEgressRequest_Egress: + roomName = v.Egress.RoomName } if roomName != "" { @@ -254,13 +283,6 @@ func (s *egressLauncher) StartEgress(ctx context.Context, req *rpc.StartEgressRe return info, nil } -func (s *egressLauncher) StopEgress(ctx context.Context, req *livekit.StopEgressRequest) (*livekit.EgressInfo, error) { - if s.client == nil { - return nil, ErrEgressNotConnected - } - return s.client.StopEgress(ctx, req.EgressId, req) -} - type LayoutMetadata struct { Layout string `json:"layout"` } @@ -372,6 +394,9 @@ func (s *EgressService) StopEgress(ctx context.Context, req *livekit.StopEgressR return info, nil } -func (s *EgressService) StartEgress(ctx context.Context, req *livekit.StartEgressRequest) (*livekit.EgressInfo, error) { - return nil, errors.New("not implemented") +func (s *egressLauncher) StopEgress(ctx context.Context, req *livekit.StopEgressRequest) (*livekit.EgressInfo, error) { + if s.client == nil { + return nil, ErrEgressNotConnected + } + return s.client.StopEgress(ctx, req.EgressId, req) } diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 5b376a490..3e497c5aa 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -23,6 +23,7 @@ import ( "github.com/pion/rtp" "github.com/pion/webrtc/v4" + "github.com/livekit/livekit-server/pkg/sfu/flexfec" sutils "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/mediatransportutil/pkg/bucket" "github.com/livekit/mediatransportutil/pkg/twcc" @@ -72,6 +73,12 @@ type Buffer struct { primaryBufferForRTX *Buffer rtxPktBuf []byte + + primaryBufferForFEC *Buffer + fecSSRC uint32 + fecDecoder *flexfec.Decoder + fecPktBuf []byte + onFECRecovery func(recovered int, received int, discarded int, bytesReceived int) } func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer { @@ -126,6 +133,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili b.pPackets = nil b.isBound = true + b.maybeCreateFECDecoderLocked() b.Unlock() if len(rtcpPackets) != 0 { @@ -179,6 +187,19 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { return } + // handle FlexFEC packet + if pb := b.primaryBufferForFEC; pb != nil { + b.Unlock() + + // skip padding only packets + if rtpPacket.Padding && len(rtpPacket.Payload) == 0 { + return + } + + pb.writeFEC(&rtpPacket, now) + return + } + if !b.isBound { packet := make([]byte, len(pkt)) copy(packet, pkt) @@ -203,6 +224,11 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { } rtcpPackets := b.calc(pkt, &rtpPacket, now, false, false) + if b.fecDecoder != nil { + // feed media into the FEC decoder, a media arrival can complete a + // previously unrecoverable FEC window + b.feedFECLocked(&rtpPacket, now) + } b.Unlock() if len(rtcpPackets) != 0 { @@ -272,6 +298,148 @@ func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) { } b.calc(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, false, true) + if b.fecDecoder != nil { + b.feedFECLocked(&repairedPkt, arrivalTime) + } +} + +func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) { + b.Lock() + b.primaryBufferForFEC = primaryBuffer + pkts := b.pPackets + b.pPackets = nil + ssrc := b.BufferBase.SSRC() + b.Unlock() + + // let the primary know the repair stream SSRC so its decoder starts + // filling with media before the first FEC packet shows up + primaryBuffer.setFECSSRC(ssrc) + + for _, pp := range pkts { + var rtpPacket rtp.Packet + err := rtpPacket.Unmarshal(pp.packet) + if err != nil { + continue + } + if rtpPacket.Padding && len(rtpPacket.Payload) == 0 { + continue + } + primaryBuffer.writeFEC(&rtpPacket, pp.arrivalTime) + } +} + +func (b *Buffer) setFECSSRC(ssrc uint32) { + b.Lock() + b.fecSSRC = ssrc + b.maybeCreateFECDecoderLocked() + b.Unlock() +} + +// maybeCreateFECDecoderLocked creates the FEC decoder as soon as the repair +// stream SSRC is known and the buffer is bound with a negotiated flexfec +// payload type. Eager creation lets the decoder track media packets before +// the first FEC packet arrives, otherwise the leading protection windows +// would be unrecoverable. +func (b *Buffer) maybeCreateFECDecoderLocked() { + if b.fecDecoder != nil || b.fecSSRC == 0 || !b.isBound || b.fecPayloadType == 0 { + return + } + + b.fecDecoder = flexfec.NewDecoder(b.fecSSRC, b.BufferBase.SSRC(), b.logger) + b.logger.Debugw("flexfec decoder created", "fecSSRC", b.fecSSRC, "mediaSSRC", b.BufferBase.SSRC()) +} + +// OnFECRecovery is called with counter deltas whenever FEC packets are +// processed: recovered media packets, FEC packets received and FEC packets +// discarded since the previous callback. +func (b *Buffer) OnFECRecovery(fn func(recovered int, received int, discarded int, bytesReceived int)) { + b.Lock() + b.onFECRecovery = fn + b.Unlock() +} + +// FECDecoderStats returns cumulative FlexFEC decode counters of the buffer. +func (b *Buffer) FECDecoderStats() flexfec.DecoderStats { + b.RLock() + defer b.RUnlock() + + if b.fecDecoder == nil { + return flexfec.DecoderStats{} + } + return b.fecDecoder.Stats() +} + +// writeFEC handles a packet of the coupled FlexFEC repair stream, recovered +// media packets are injected into the regular packet pipeline. +func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) { + b.Lock() + if !b.isBound { + b.Unlock() + return + } + + // the FEC stream is never bound in pion, run its TWCC accounting here so + // publisher send side BWE sees the FEC packets acked. The repair stream + // shares the media m-line, extension ids match the primary stream. + if b.twcc != nil && b.twccExtID != 0 { + if ext := fecPkt.GetExtension(b.twccExtID); ext != nil { + b.twcc.Push(fecPkt.SSRC, binary.BigEndian.Uint16(ext[0:2]), arrivalTime, fecPkt.Marker) + } + } + + if b.fecPayloadType == 0 || fecPkt.PayloadType != b.fecPayloadType { + b.logger.Debugw("unexpected fec payload type", "expected", b.fecPayloadType, "actual", fecPkt.PayloadType) + b.Unlock() + return + } + + if b.fecDecoder == nil { + // normally created when the pair is declared, fall back to the + // observed repair stream SSRC + b.fecSSRC = fecPkt.SSRC + b.maybeCreateFECDecoderLocked() + if b.fecDecoder == nil { + b.Unlock() + return + } + } + + b.feedFECLocked(fecPkt, arrivalTime) + b.Unlock() +} + +// feedFECLocked runs a media or FEC packet through the FEC decoder and +// injects recovered packets into the packet pipeline. Must be called with the +// buffer lock held and a non-nil decoder. +func (b *Buffer) feedFECLocked(pkt *rtp.Packet, arrivalTime int64) { + statsBefore := b.fecDecoder.Stats() + recovered := b.fecDecoder.DecodeFec(pkt) + + if b.fecPktBuf == nil { + b.fecPktBuf = make([]byte, bucket.RTPMaxPktSize) + } + for _, rp := range recovered { + n, err := rp.MarshalTo(b.fecPktBuf) + if err != nil { + b.logger.Warnw("could not marshal fec recovered packet", err, "ssrc", b.BufferBase.SSRC(), "sn", rp.SequenceNumber) + continue + } + + // recovered packets flow through the regular pipeline: they are + // forwarded downstream and stop NACKs for the lost sequence numbers. + // they do not re-enter the decoder, it already has them in its window. + b.calc(b.fecPktBuf[:n], rp, arrivalTime, false, true) + } + + if cb := b.onFECRecovery; cb != nil { + statsAfter := b.fecDecoder.Stats() + received := int(statsAfter.FECPacketsReceived - statsBefore.FECPacketsReceived) + discarded := int(statsAfter.FECPacketsDiscarded - statsBefore.FECPacketsDiscarded) + bytesReceived := int(statsAfter.FECBytesReceived - statsBefore.FECBytesReceived) + if len(recovered) > 0 || received > 0 || discarded > 0 { + cb(len(recovered), received, discarded, bytesReceived) + } + } } func (b *Buffer) Read(buff []byte) (n int, err error) { diff --git a/pkg/sfu/buffer/buffer_base.go b/pkg/sfu/buffer/buffer_base.go index b94326cff..12a264c48 100644 --- a/pkg/sfu/buffer/buffer_base.go +++ b/pkg/sfu/buffer/buffer_base.go @@ -168,6 +168,7 @@ type BufferBase struct { rtpParameters webrtc.RTPParameters payloadType uint8 rtxPayloadType uint8 + fecPayloadType uint8 snRangeMap *utils.RangeMap[uint64, uint64] @@ -312,6 +313,14 @@ func (b *BufferBase) BindLocked(rtpParameters webrtc.RTPParameters, codec webrtc } } + // find FlexFEC payload type, the repair stream is codec agnostic (no apt) + for _, codec := range rtpParameters.Codecs { + if strings.EqualFold(codec.MimeType, webrtc.MimeTypeFlexFEC03) { + b.fecPayloadType = uint8(codec.PayloadType) + break + } + } + for _, ext := range rtpParameters.HeaderExtensions { switch ext.URI { case dd.ExtensionURI: diff --git a/pkg/sfu/buffer/buffer_fec_test.go b/pkg/sfu/buffer/buffer_fec_test.go new file mode 100644 index 000000000..93b3395ad --- /dev/null +++ b/pkg/sfu/buffer/buffer_fec_test.go @@ -0,0 +1,324 @@ +// 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 buffer + +import ( + "math/rand" + "testing" + + pionflexfec "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtp" + "github.com/pion/transport/v4/packetio" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + fecTestMediaSSRC = uint32(0x11111111) + fecTestFECSSRC = uint32(0x22222222) + fecTestFECPT = uint8(115) +) + +var flexfecCodec = webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeFlexFEC03, + ClockRate: 90000, + SDPFmtpLine: "repair-window=10000000", + }, + PayloadType: webrtc.PayloadType(fecTestFECPT), +} + +func fecTestMediaPackets(t *testing.T, baseSN uint16, count int) []rtp.Packet { + t.Helper() + rng := rand.New(rand.NewSource(int64(baseSN))) + pkts := make([]rtp.Packet, 0, count) + for i := 0; i < count; i++ { + payload := make([]byte, 50+rng.Intn(200)) + rng.Read(payload) + // valid VP8 payload descriptor (S=1, no extensions) so the video + // packet processing in the buffer accepts the packet + payload[0] = 0x10 + sn := baseSN + uint16(i) + pkts = append(pkts, rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: uint8(vp8Codec.PayloadType), + SequenceNumber: sn, + // derive timestamp from the sequence number so windows + // generated separately stay monotonic + Timestamp: 90000 + 3000*uint32(sn), + SSRC: fecTestMediaSSRC, + Marker: i == count-1, + }, + Payload: payload, + }) + } + return pkts +} + +func bindFECTestBuffer(t *testing.T, buff *Buffer) { + t.Helper() + buff.codecType = webrtc.RTPCodecTypeVideo + require.NoError(t, buff.Bind(webrtc.RTPParameters{ + Codecs: []webrtc.RTPCodecParameters{vp8Codec, flexfecCodec}, + }, vp8Codec.RTPCodecCapability, 0)) +} + +func writePacket(t *testing.T, buff *Buffer, pkt *rtp.Packet) { + t.Helper() + raw, err := pkt.Marshal() + require.NoError(t, err) + _, err = buff.Write(raw) + require.NoError(t, err) +} + +// readExtSequenceNumbers drains count ExtPackets and returns sequence number +// -> extended sequence number of everything seen. +func readExtSequenceNumbers(t *testing.T, buff *Buffer, count int) map[uint16]uint64 { + t.Helper() + seen := make(map[uint16]uint64, count) + var buf [1500]byte + for i := 0; i < count; i++ { + extPkt, err := buff.ReadExtended(buf[:]) + require.NoError(t, err) + require.NotNil(t, extPkt) + seen[extPkt.Packet.SequenceNumber] = extPkt.ExtSequenceNumber + } + return seen +} + +// requireRecoveredInBucket asserts that the dropped packet was placed into +// the buffer's bucket (where downstream NACKs are served from), matching the +// behavior of RTX repaired packets. +func requireRecoveredInBucket(t *testing.T, buff *Buffer, dropped *rtp.Packet, extSNBySN map[uint16]uint64, refSN uint16) { + t.Helper() + refExtSN, ok := extSNBySN[refSN] + require.True(t, ok, "reference sequence number %d not seen", refSN) + droppedExtSN := refExtSN + uint64(dropped.SequenceNumber-refSN) + + var buf [1500]byte + n, err := buff.GetPacket(buf[:], droppedExtSN) + require.NoError(t, err, "recovered packet not found in bucket") + + var pkt rtp.Packet + require.NoError(t, pkt.Unmarshal(buf[:n])) + require.Equal(t, dropped.SequenceNumber, pkt.SequenceNumber) + assert.Equal(t, dropped.Timestamp, pkt.Timestamp) + assert.Equal(t, dropped.Payload, pkt.Payload) +} + +func TestBufferFECRecoversDroppedPacket(t *testing.T) { + factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory() + + primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer) + fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) + factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) + + bindFECTestBuffer(t, primary) + + var recoveredDelta, receivedDelta int + primary.OnFECRecovery(func(recovered int, received int, discarded int, bytesReceived int) { + recoveredDelta += recovered + receivedDelta += received + }) + + media := fecTestMediaPackets(t, 100, 10) + encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) + fecPackets := encoder.EncodeFec(media, 2) + require.NotEmpty(t, fecPackets) + + const droppedIdx = 3 + for i := range media { + if i == droppedIdx { + continue + } + writePacket(t, primary, &media[i]) + } + for i := range fecPackets { + writePacket(t, fecBuff, &fecPackets[i]) + } + + stats := primary.FECDecoderStats() + assert.EqualValues(t, len(fecPackets), stats.FECPacketsReceived) + assert.EqualValues(t, 1, stats.PacketsRecovered) + assert.EqualValues(t, 0, stats.FECPacketsDiscarded) + assert.Equal(t, 1, recoveredDelta) + assert.Equal(t, len(fecPackets), receivedDelta) + + // the 9 received packets flow through the ext packet pipeline, the + // recovered one fills the bucket like an RTX repair + extSNBySN := readExtSequenceNumbers(t, primary, len(media)-1) + requireRecoveredInBucket(t, primary, &media[droppedIdx], extSNBySN, media[0].SequenceNumber) +} + +func TestBufferFECPairAfterPackets(t *testing.T) { + // FEC packets arriving before the ssrc-group is known are queued as + // pending and replayed when the pair is established. Media seen before + // the pairing is not in the decoder window (cold start), so the first + // window is not recoverable, subsequent windows are. + factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory() + + primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer) + bindFECTestBuffer(t, primary) + + encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) + media := fecTestMediaPackets(t, 200, 10) + fecPackets := encoder.EncodeFec(media, 2) + require.NotEmpty(t, fecPackets) + + for i := range media { + if i == 5 { + continue + } + writePacket(t, primary, &media[i]) + } + + // fec buffer created by first packet arrival, before the pair is declared + fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) + for i := range fecPackets { + writePacket(t, fecBuff, &fecPackets[i]) + } + + stats := primary.FECDecoderStats() + require.EqualValues(t, 0, stats.FECPacketsReceived) + + factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) + + // pending FEC was replayed into the decoder, no recovery possible for the + // cold-start window + stats = primary.FECDecoderStats() + assert.EqualValues(t, len(fecPackets), stats.FECPacketsReceived) + assert.EqualValues(t, 0, stats.PacketsRecovered) + + // the next window recovers normally + media2 := fecTestMediaPackets(t, 210, 10) + fecPackets2 := encoder.EncodeFec(media2, 2) + require.NotEmpty(t, fecPackets2) + + const droppedIdx = 4 + for i := range media2 { + if i == droppedIdx { + continue + } + writePacket(t, primary, &media2[i]) + } + for i := range fecPackets2 { + writePacket(t, fecBuff, &fecPackets2[i]) + } + + stats = primary.FECDecoderStats() + assert.EqualValues(t, 1, stats.PacketsRecovered) + + extSNBySN := readExtSequenceNumbers(t, primary, len(media)-1+len(media2)-1) + requireRecoveredInBucket(t, primary, &media2[droppedIdx], extSNBySN, media2[0].SequenceNumber) +} + +func TestBufferFECCoupledBeforeBuffersExist(t *testing.T) { + // pair declared first (from SDP), buffers created later on first packet + factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory() + factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) + + primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer) + fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) + bindFECTestBuffer(t, primary) + + media := fecTestMediaPackets(t, 300, 5) + encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) + fecPackets := encoder.EncodeFec(media, 1) + require.NotEmpty(t, fecPackets) + + const droppedIdx = 2 + for i := range media { + if i == droppedIdx { + continue + } + writePacket(t, primary, &media[i]) + } + for i := range fecPackets { + writePacket(t, fecBuff, &fecPackets[i]) + } + + stats := primary.FECDecoderStats() + assert.EqualValues(t, 1, stats.PacketsRecovered) + + extSNBySN := readExtSequenceNumbers(t, primary, len(media)-1) + requireRecoveredInBucket(t, primary, &media[droppedIdx], extSNBySN, media[0].SequenceNumber) +} + +func TestBufferFECIgnoresUnexpectedPayloadType(t *testing.T) { + factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory() + + primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer) + fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) + factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) + + // bound without flexfec in negotiated codecs + primary.codecType = webrtc.RTPCodecTypeVideo + require.NoError(t, primary.Bind(webrtc.RTPParameters{ + Codecs: []webrtc.RTPCodecParameters{vp8Codec}, + }, vp8Codec.RTPCodecCapability, 0)) + + media := fecTestMediaPackets(t, 400, 5) + encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) + fecPackets := encoder.EncodeFec(media, 1) + require.NotEmpty(t, fecPackets) + + for i := range media { + writePacket(t, primary, &media[i]) + } + for i := range fecPackets { + writePacket(t, fecBuff, &fecPackets[i]) + } + + // no flexfec payload type negotiated, decoder must not be created + stats := primary.FECDecoderStats() + assert.EqualValues(t, 0, stats.FECPacketsReceived) + assert.EqualValues(t, 0, stats.PacketsRecovered) +} + +func TestBufferFECNACKSuppression(t *testing.T) { + // a recovered packet must clear the pending NACK for its sequence number + factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory() + + primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer) + fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) + factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) + bindFECTestBuffer(t, primary) + + media := fecTestMediaPackets(t, 700, 10) + encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) + fecPackets := encoder.EncodeFec(media, 2) + require.NotEmpty(t, fecPackets) + + const droppedIdx = 6 + for i := range media { + if i == droppedIdx { + continue + } + writePacket(t, primary, &media[i]) + } + + // the only gap is the dropped packet, exactly one queued NACK + require.NotNil(t, primary.nacker) + require.Len(t, primary.nacker.Nacks(), 1, "expected queued NACK for dropped packet") + + for i := range fecPackets { + writePacket(t, fecBuff, &fecPackets[i]) + } + require.EqualValues(t, 1, primary.FECDecoderStats().PacketsRecovered) + + require.Empty(t, primary.nacker.Nacks(), "NACK for recovered packet not suppressed") +} diff --git a/pkg/sfu/buffer/factory.go b/pkg/sfu/buffer/factory.go index 4a73ab057..802909e8a 100644 --- a/pkg/sfu/buffer/factory.go +++ b/pkg/sfu/buffer/factory.go @@ -40,6 +40,7 @@ func (f *FactoryOfBufferFactory) CreateBufferFactory() *Factory { rtpBuffers: make(map[uint32]*Buffer), rtcpReaders: make(map[uint32]*RTCPReader), rtxPair: make(map[uint32]uint32), + fecPair: make(map[uint32]uint32), } } @@ -50,6 +51,7 @@ type Factory struct { rtpBuffers map[uint32]*Buffer rtcpReaders map[uint32]*RTCPReader rtxPair map[uint32]uint32 // repair -> base + fecPair map[uint32]uint32 // fec -> base } func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser { @@ -89,10 +91,26 @@ func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io break } } + for fec, base := range f.fecPair { + if fec == ssrc { + baseBuffer, ok := f.rtpBuffers[base] + if ok { + buffer.SetPrimaryBufferForFEC(baseBuffer) + } + break + } else if base == ssrc { + fecBuffer, ok := f.rtpBuffers[fec] + if ok { + fecBuffer.SetPrimaryBufferForFEC(buffer) + } + break + } + } buffer.OnClose(func() { f.Lock() delete(f.rtpBuffers, ssrc) delete(f.rtxPair, ssrc) + delete(f.fecPair, ssrc) f.Unlock() }) return buffer @@ -132,3 +150,15 @@ func (f *Factory) SetRTXPair(repair, base uint32, rsid string) { } } } + +func (f *Factory) SetFECPair(fec, base uint32) { + f.Lock() + fecBuffer, baseBuffer := f.rtpBuffers[fec], f.rtpBuffers[base] + if fecBuffer == nil || baseBuffer == nil { + f.fecPair[fec] = base + } + f.Unlock() + if fecBuffer != nil && baseBuffer != nil { + fecBuffer.SetPrimaryBufferForFEC(baseBuffer) + } +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 38cf31569..9f2f52ded 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -46,6 +46,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/packettrailer" act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime" dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor" + "github.com/livekit/livekit-server/pkg/telemetry/prometheus" pd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/playoutdelay" "github.com/livekit/livekit-server/pkg/sfu/rtpstats" "github.com/livekit/livekit-server/pkg/sfu/utils" @@ -295,6 +296,13 @@ var _ TrackSender = (*DownTrack)(nil) type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) +// FlexFECParams configures FlexFEC-03 generation toward the subscriber. +type FlexFECParams struct { + Enabled bool + NumMediaPackets uint32 + NumFECPackets uint32 +} + type DownTrackParams struct { Codecs []webrtc.RTPCodecParameters IsEncrypted bool @@ -313,6 +321,7 @@ type DownTrackParams struct { SupportsCodecChange bool StripPacketTrailer bool Listener DownTrackListener + FlexFEC FlexFECParams } // DownTrack implements webrtc.TrackLocal, is the track used to write packets @@ -329,8 +338,11 @@ type DownTrack struct { kind webrtc.RTPCodecType ssrc uint32 ssrcRTX uint32 + ssrcFEC uint32 payloadType atomic.Uint32 payloadTypeRTX atomic.Uint32 + payloadTypeFEC atomic.Uint32 + fecWriter atomic.Pointer[fecWriter] sequencer *sequencer rtxSequenceNumber atomic.Uint64 @@ -620,14 +632,24 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, d.ssrc = uint32(t.SSRC()) d.ssrcRTX = uint32(t.SSRCRetransmission()) + d.ssrcFEC = uint32(t.SSRCForwardErrorCorrection()) d.payloadType.Store(uint32(codec.PayloadType)) d.payloadTypeRTX.Store(uint32(utils.FindRTXPayloadType(codec.PayloadType, d.negotiatedCodecParameters))) + d.payloadTypeFEC.Store(uint32(utils.FindFlexFECPayloadType(d.negotiatedCodecParameters))) + d.maybeCreateFECWriter() logFields = append( logFields, "payloadType", d.payloadType.Load(), "payloadTypeRTX", d.payloadTypeRTX.Load(), "codecParameters", d.negotiatedCodecParameters, ) + if fw := d.fecWriter.Load(); fw != nil { + logFields = append( + logFields, + "ssrcFEC", d.ssrcFEC, + "payloadTypeFEC", d.payloadTypeFEC.Load(), + ) + } d.params.Logger.Debugw("DownTrack.Bind", logFields...) d.writeStream = t.WriteStream() @@ -1195,6 +1217,11 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { Pool: PacketFactory, PoolEntity: poolEntity, } + if d.fecWriter.Load() != nil { + // protect the packet as sent on the wire, the pacer patched header + // included + pacerPacket.OnSent = d.onMediaPacketSentForFEC + } d.pacer.Enqueue(pacerPacket) if extPkt.IsKeyFrame { @@ -1333,6 +1360,11 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa TransportWideExtID: uint8(d.transportWideExtID), WriteStream: d.writeStream, } + if d.fecWriter.Load() != nil { + // padding probes consume media sequence numbers, keep them in + // the FEC window so probing does not break its continuity + pacerPacket.OnSent = d.onMediaPacketSentForFEC + } d.pacer.Enqueue(pacerPacket) bytesSent += hdrSize + payloadSize @@ -2448,6 +2480,73 @@ func (d *DownTrack) addDummyExtensions(hdr *rtp.Header) { } } +// maybeCreateFECWriter starts FlexFEC generation when enabled and the +// subscriber negotiated a FEC SSRC and the flexfec-03 codec. +func (d *DownTrack) maybeCreateFECWriter() { + if !d.params.FlexFEC.Enabled || + d.kind != webrtc.RTPCodecTypeVideo || + d.ssrcFEC == 0 || + d.payloadTypeFEC.Load() == 0 || + d.fecWriter.Load() != nil { + return + } + + d.fecWriter.Store(newFECWriter( + d.ssrcFEC, + uint8(d.payloadTypeFEC.Load()), + d.params.FlexFEC.NumMediaPackets, + d.params.FlexFEC.NumFECPackets, + d.params.Logger, + )) +} + +// onMediaPacketSentForFEC is invoked by the pacer with the final on-wire +// header and payload of every forwarded media packet. Completed protection +// windows yield FEC packets which are queued behind the media they protect. +func (d *DownTrack) onMediaPacketSentForFEC(hdr *rtp.Header, payload []byte) { + fw := d.fecWriter.Load() + if fw == nil { + return + } + + fecPackets := fw.add(hdr, payload) + if len(fecPackets) == 0 { + return + } + + bytesSent := uint64(0) + for i := range fecPackets { + fecHdr := fecPackets[i].Header + d.addDummyExtensions(&fecHdr) + headerSize := fecHdr.MarshalSize() + + pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet) + *pacerPacket = pacer.Packet{ + Header: &fecHdr, + HeaderSize: headerSize, + Payload: fecPackets[i].Payload, + ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()), + AbsSendTimeExtID: uint8(d.absSendTimeExtID), + TransportWideExtID: uint8(d.transportWideExtID), + WriteStream: d.writeStream, + } + bytesSent += uint64(headerSize + len(fecPackets[i].Payload)) + d.pacer.Enqueue(pacerPacket) + } + + prometheus.RecordFECDownstreamSent(len(fecPackets), bytesSent) +} + +// FECWriterStats returns cumulative downstream FlexFEC generation counters, +// the second return indicates whether FEC generation is active. +func (d *DownTrack) FECWriterStats() (FECWriterStats, bool) { + fw := d.fecWriter.Load() + if fw == nil { + return FECWriterStats{}, false + } + return fw.Stats(), true +} + func (d *DownTrack) getTranslatedPayloadType(srcPT uint8) uint8 { // send primary codec to subscriber if the publisher sent primary codec when red is negotiated, // this will happen when the payload is too large to encode into red payload (exceeds mtu). @@ -2470,6 +2569,17 @@ func (d *DownTrack) DebugInfo() map[string]any { stats["PacketCount"] = senderReport.PacketCount } + if fecStats, ok := d.FECWriterStats(); ok { + stats["FlexFEC"] = map[string]any{ + "SSRC": d.ssrcFEC, + "PayloadType": d.payloadTypeFEC.Load(), + "PacketsSent": fecStats.PacketsSent, + "BytesSent": fecStats.BytesSent, + "PartialWindows": fecStats.PartialWindows, + "DiscardedSingles": fecStats.DiscardedSingles, + } + } + return map[string]any{ "SubscriberID": d.params.SubID, "TrackID": d.id, diff --git a/pkg/sfu/downtrack_fec.go b/pkg/sfu/downtrack_fec.go new file mode 100644 index 000000000..db0fd2ea3 --- /dev/null +++ b/pkg/sfu/downtrack_fec.go @@ -0,0 +1,134 @@ +// 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 ( + "sync" + + pionflexfec "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtp" + + "github.com/livekit/protocol/logger" +) + +// FECWriterStats accumulates downstream FlexFEC generation counters. +type FECWriterStats struct { + PacketsSent uint64 + BytesSent uint64 + // windows flushed early because of a sequence number discontinuity + // (unrecovered upstream losses propagate into the forwarded stream, + // probe padding can interleave out of order) + PartialWindows uint64 + // single packets dropped from the window on a discontinuity, too small + // to protect + DiscardedSingles uint64 +} + +// fecWriter generates FlexFEC-03 repair packets for media sent on a +// DownTrack. Media packets are fed with their final on-wire header and +// payload (after the pacer patched extensions) so the generated FEC matches +// the bytes the subscriber receives. The flexfec encoder requires gap free +// in-order windows, on a discontinuity the consecutive run collected so far +// is protected as a partial window (with the FEC count scaled down to keep +// the overhead ratio) instead of being discarded, gaps are common when +// upstream losses could not be recovered. +type fecWriter struct { + logger logger.Logger + ssrc uint32 + payloadType uint8 + numMediaPackets uint32 + numFECPackets uint32 + + lock sync.Mutex + encoder *pionflexfec.FlexEncoder03 + window []rtp.Packet + nextSN uint16 + haveNext bool + stats FECWriterStats +} + +func newFECWriter(ssrc uint32, payloadType uint8, numMediaPackets uint32, numFECPackets uint32, logger logger.Logger) *fecWriter { + return &fecWriter{ + logger: logger, + ssrc: ssrc, + payloadType: payloadType, + numMediaPackets: numMediaPackets, + numFECPackets: numFECPackets, + encoder: pionflexfec.NewFlexEncoder03(payloadType, ssrc), + window: make([]rtp.Packet, 0, numMediaPackets), + } +} + +// add ingests a sent media packet and returns FEC packets to transmit when a +// protection window completes. The header and payload are copied, callers +// may reuse their memory. +func (w *fecWriter) add(hdr *rtp.Header, payload []byte) []rtp.Packet { + w.lock.Lock() + defer w.lock.Unlock() + + var fecPackets []rtp.Packet + if w.haveNext && hdr.SequenceNumber != w.nextSN { + fecPackets = w.flushLocked() + if fecPackets != nil { + w.stats.PartialWindows++ + } + } + w.nextSN = hdr.SequenceNumber + 1 + w.haveNext = true + + w.window = append(w.window, rtp.Packet{ + Header: hdr.Clone(), + Payload: append([]byte(nil), payload...), + }) + if len(w.window) >= int(w.numMediaPackets) { + fecPackets = append(fecPackets, w.flushLocked()...) + } + return fecPackets +} + +// flushLocked protects the current window and resets it. Windows of a single +// packet are dropped, a one packet XOR is just a retransmission. +func (w *fecWriter) flushLocked() []rtp.Packet { + defer func() { + w.window = w.window[:0] + }() + + if len(w.window) < 2 { + if len(w.window) == 1 { + w.stats.DiscardedSingles++ + } + return nil + } + + // scale the FEC count to the window size to keep the overhead ratio + numFEC := (w.numFECPackets*uint32(len(w.window)) + w.numMediaPackets - 1) / w.numMediaPackets + if numFEC == 0 { + numFEC = 1 + } + + fecPackets := w.encoder.EncodeFec(w.window, numFEC) + for i := range fecPackets { + w.stats.PacketsSent++ + w.stats.BytesSent += uint64(len(fecPackets[i].Payload)) + } + return fecPackets +} + +func (w *fecWriter) Stats() FECWriterStats { + w.lock.Lock() + defer w.lock.Unlock() + + return w.stats +} diff --git a/pkg/sfu/downtrack_fec_test.go b/pkg/sfu/downtrack_fec_test.go new file mode 100644 index 000000000..3d7d30ba8 --- /dev/null +++ b/pkg/sfu/downtrack_fec_test.go @@ -0,0 +1,206 @@ +// 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 ( + "math/rand" + "testing" + + "github.com/pion/rtp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/sfu/flexfec" + "github.com/livekit/protocol/logger" +) + +const ( + fecWriterTestMediaSSRC = uint32(0xaaaa1111) + fecWriterTestFECSSRC = uint32(0xbbbb2222) + fecWriterTestFECPT = uint8(115) +) + +func fecWriterTestPacket(rng *rand.Rand, sn uint16) (*rtp.Header, []byte) { + payload := make([]byte, 100+rng.Intn(400)) + rng.Read(payload) + return &rtp.Header{ + Version: 2, + PayloadType: 96, + SequenceNumber: sn, + Timestamp: 3000 * uint32(sn), + SSRC: fecWriterTestMediaSSRC, + Marker: sn%5 == 4, + }, payload +} + +func TestFECWriterEmitsOnWindowCompletion(t *testing.T) { + w := newFECWriter(fecWriterTestFECSSRC, fecWriterTestFECPT, 5, 2, logger.GetLogger()) + rng := rand.New(rand.NewSource(1)) + + var emitted []rtp.Packet + for sn := uint16(100); sn < 110; sn++ { + hdr, payload := fecWriterTestPacket(rng, sn) + emitted = append(emitted, w.add(hdr, payload)...) + } + + // two complete windows of 5, each yielding 2 FEC packets + require.Len(t, emitted, 4) + for _, fec := range emitted { + assert.Equal(t, fecWriterTestFECSSRC, fec.SSRC) + assert.Equal(t, fecWriterTestFECPT, fec.PayloadType) + } + + stats := w.Stats() + assert.EqualValues(t, 4, stats.PacketsSent) + assert.NotZero(t, stats.BytesSent) + assert.EqualValues(t, 0, stats.PartialWindows) +} + +func TestFECWriterResetsOnDiscontinuity(t *testing.T) { + w := newFECWriter(fecWriterTestFECSSRC, fecWriterTestFECPT, 5, 1, logger.GetLogger()) + rng := rand.New(rand.NewSource(2)) + + // 3 packets, then a gap (e.g. an unrecovered upstream loss), then a + // full window. The partial run before the gap is protected too. + var emitted []rtp.Packet + for _, sn := range []uint16{100, 101, 102, 110, 111, 112, 113, 114} { + hdr, payload := fecWriterTestPacket(rng, sn) + emitted = append(emitted, w.add(hdr, payload)...) + } + + // one partial window (100-102) + one full window (110-114) + require.Len(t, emitted, 2) + stats := w.Stats() + assert.EqualValues(t, 1, stats.PartialWindows) + assert.EqualValues(t, 2, stats.PacketsSent) +} + +func TestFECWriterDropsSinglePacketWindows(t *testing.T) { + w := newFECWriter(fecWriterTestFECSSRC, fecWriterTestFECPT, 5, 1, logger.GetLogger()) + rng := rand.New(rand.NewSource(7)) + + var emitted []rtp.Packet + for _, sn := range []uint16{100, 110, 120, 121} { + hdr, payload := fecWriterTestPacket(rng, sn) + emitted = append(emitted, w.add(hdr, payload)...) + } + require.Empty(t, emitted) + stats := w.Stats() + assert.EqualValues(t, 2, stats.DiscardedSingles) + assert.EqualValues(t, 0, stats.PacketsSent) +} + +func TestFECWriterPartialWindowRecovers(t *testing.T) { + // a partial window flushed on a gap must still produce usable FEC + w := newFECWriter(fecWriterTestFECSSRC, fecWriterTestFECPT, 10, 2, logger.GetLogger()) + rng := rand.New(rand.NewSource(8)) + + originals := make([]rtp.Packet, 0, 4) + var emitted []rtp.Packet + for _, sn := range []uint16{300, 301, 302, 303} { + hdr, payload := fecWriterTestPacket(rng, sn) + originals = append(originals, rtp.Packet{Header: hdr.Clone(), Payload: append([]byte(nil), payload...)}) + emitted = append(emitted, w.add(hdr, payload)...) + } + // gap triggers the partial flush + hdr, payload := fecWriterTestPacket(rng, 310) + emitted = append(emitted, w.add(hdr, payload)...) + require.NotEmpty(t, emitted) + + decoder := flexfec.NewDecoder(fecWriterTestFECSSRC, fecWriterTestMediaSSRC, logger.GetLogger()) + for i := range originals { + if i == 2 { + continue + } + require.Empty(t, decoder.DecodeFec(&originals[i])) + } + recovered := decoder.DecodeFec(&emitted[0]) + require.Len(t, recovered, 1) + assert.Equal(t, originals[2].SequenceNumber, recovered[0].SequenceNumber) + assert.Equal(t, originals[2].Payload, recovered[0].Payload) +} + +func TestFECWriterCopiesCallerMemory(t *testing.T) { + w := newFECWriter(fecWriterTestFECSSRC, fecWriterTestFECPT, 3, 1, logger.GetLogger()) + rng := rand.New(rand.NewSource(3)) + + originals := make([]rtp.Packet, 0, 3) + scratchPayload := make([]byte, 1500) + var emitted []rtp.Packet + for sn := uint16(50); sn < 53; sn++ { + hdr, payload := fecWriterTestPacket(rng, sn) + originals = append(originals, rtp.Packet{Header: *hdr, Payload: append([]byte(nil), payload...)}) + + // hand the writer reused scratch memory + copy(scratchPayload, payload) + emitted = append(emitted, w.add(hdr, scratchPayload[:len(payload)])...) + for i := range scratchPayload { + scratchPayload[i] = 0xee + } + } + require.Len(t, emitted, 1) + + // the FEC packet must recover the original bytes, proving the writer + // copied rather than aliased the scratch memory + decoder := flexfec.NewDecoder(fecWriterTestFECSSRC, fecWriterTestMediaSSRC, logger.GetLogger()) + for i := range originals { + if i == 1 { + continue + } + require.Empty(t, decoder.DecodeFec(&originals[i])) + } + recovered := decoder.DecodeFec(&emitted[0]) + require.Len(t, recovered, 1) + assert.Equal(t, originals[1].SequenceNumber, recovered[0].SequenceNumber) + assert.Equal(t, originals[1].Payload, recovered[0].Payload) + assert.Equal(t, originals[1].Timestamp, recovered[0].Timestamp) +} + +func TestFECWriterRoundTripWithExtensions(t *testing.T) { + // simulate the wire path: headers carry abs-send-time/transport-cc + // extensions like the pacer emits them + w := newFECWriter(fecWriterTestFECSSRC, fecWriterTestFECPT, 5, 1, logger.GetLogger()) + rng := rand.New(rand.NewSource(4)) + + wirePackets := make([]rtp.Packet, 0, 5) + var emitted []rtp.Packet + for sn := uint16(800); sn < 805; sn++ { + hdr, payload := fecWriterTestPacket(rng, sn) + require.NoError(t, hdr.SetExtension(2, []byte{0x01, 0x02, 0x03})) + twcc := []byte{byte(sn >> 8), byte(sn)} + require.NoError(t, hdr.SetExtension(3, twcc)) + + wirePackets = append(wirePackets, rtp.Packet{Header: hdr.Clone(), Payload: append([]byte(nil), payload...)}) + emitted = append(emitted, w.add(hdr, payload)...) + } + require.Len(t, emitted, 1) + + const droppedIdx = 2 + decoder := flexfec.NewDecoder(fecWriterTestFECSSRC, fecWriterTestMediaSSRC, logger.GetLogger()) + for i := range wirePackets { + if i == droppedIdx { + continue + } + require.Empty(t, decoder.DecodeFec(&wirePackets[i])) + } + recovered := decoder.DecodeFec(&emitted[0]) + require.Len(t, recovered, 1) + + expected, err := wirePackets[droppedIdx].Marshal() + require.NoError(t, err) + actual, err := recovered[0].Marshal() + require.NoError(t, err) + assert.Equal(t, expected, actual, "recovered wire bytes must match the original, extensions included") +} diff --git a/pkg/sfu/flexfec/decoder.go b/pkg/sfu/flexfec/decoder.go new file mode 100644 index 000000000..1d4e2415b --- /dev/null +++ b/pkg/sfu/flexfec/decoder.go @@ -0,0 +1,504 @@ +// 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 implements a FlexFEC-03 decoder used to recover RTP packets +// lost on the publisher leg before they are forwarded downstream. +// https://datatracker.ietf.org/doc/html/draft-ietf-payload-flexible-fec-scheme-03 +// +// The 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: +// - packets are deep-copied on insertion (callers reuse packet memory) +// - the media window holds stable heap pointers; the pion version keeps +// values and re-sorts them in place, which invalidates the references +// held by FEC packet state on out-of-order arrival +// - failed recoveries are not emitted as empty packets +// - usage counters for metrics +package flexfec + +import ( + "encoding/binary" + "errors" + "fmt" + "sort" + + "github.com/pion/rtp" + + "github.com/livekit/protocol/logger" +) + +var ( + errPacketTruncated = errors.New("packet truncated") + errRetransmissionBitSet = errors.New("packet with retransmission bit set not supported") + errInflexibleGeneratorMatrix = errors.New("packet with inflexible generator matrix not supported") + errMultipleSSRCProtection = errors.New("multiple ssrc protection not supported") + errLastOptionalMaskKBitSetToFalse = errors.New("k-bit of last optional mask is set to false") + errEmptyMask = errors.New("empty fec packet mask") + errUnknownProtectedSSRC = errors.New("fec is protecting unknown ssrc") +) + +const ( + // media window size that triggers the sequence gap reset check + maxMediaPackets = 100 + // maximum number of FEC packets retained + maxFECPackets = 100 + // seen/recovered media packets retained for XOR recovery + recoveredPacketsLimit = 192 +) + +// DecoderStats accumulates FEC usage counters. Snapshot via Decoder.Stats. +type DecoderStats struct { + // FEC packets fed to the decoder + FECPacketsReceived uint64 + // FEC bytes fed to the decoder (RTP payload sizes) + FECBytesReceived uint64 + // FEC packets that could not be used: parse failures, foreign protected + // SSRC, empty masks and duplicates + FECPacketsDiscarded uint64 + // media packets reconstructed from FEC + PacketsRecovered uint64 +} + +// Decoder recovers lost media packets of a single protected SSRC from a +// FlexFEC-03 repair stream. It is not safe for concurrent use; the owning +// buffer serializes access. +type Decoder struct { + logger logger.Logger + fecSSRC uint32 + protectedSSRC uint32 + recoveredPackets []*rtp.Packet + receivedFECPackets []fecPacketState + stats DecoderStats +} + +func NewDecoder(fecSSRC uint32, protectedSSRC uint32, logger logger.Logger) *Decoder { + return &Decoder{ + logger: logger, + fecSSRC: fecSSRC, + protectedSSRC: protectedSSRC, + } +} + +func (d *Decoder) Stats() DecoderStats { + return d.stats +} + +// DecodeFec ingests a packet of either the FEC stream (fecSSRC) or the +// protected media stream (protectedSSRC) and returns any media packets that +// became recoverable. Returned packets are owned by the decoder's internal +// window; callers must not mutate them. +func (d *Decoder) DecodeFec(receivedPacket *rtp.Packet) []*rtp.Packet { + if receivedPacket.SSRC == d.fecSSRC { + d.stats.FECPacketsReceived++ + d.stats.FECBytesReceived += uint64(len(receivedPacket.Payload)) + } + + // the caller reuses packet memory, keep an owned copy + pkt := clonePacket(receivedPacket) + + if len(d.recoveredPackets) >= maxMediaPackets { + backRecoveredPacket := d.recoveredPackets[len(d.recoveredPackets)-1] + if backRecoveredPacket.SSRC == pkt.SSRC { + if seqDiff(pkt.SequenceNumber, backRecoveredPacket.SequenceNumber) > uint16(maxMediaPackets) { + d.logger.Infow("flexfec: big gap in media sequence numbers - resetting buffers") + d.recoveredPackets = nil + d.receivedFECPackets = nil + } + } + } + + d.insertPacket(pkt) + + recovered := d.attemptRecovery() + d.stats.PacketsRecovered += uint64(len(recovered)) + return recovered +} + +func (d *Decoder) insertPacket(receivedPkt *rtp.Packet) { + // Discard old FEC packets such that the sequence numbers in + // `receivedFECPackets` span at most 1/2 of the sequence number space. + // This is important for keeping `receivedFECPackets` sorted, and may + // also reduce the possibility of incorrect decoding due to sequence + // number wrap-around. + if len(d.receivedFECPackets) > 0 && receivedPkt.SSRC == d.fecSSRC { + toRemove := 0 + for _, fecPkt := range d.receivedFECPackets { + if absInt(int(receivedPkt.SequenceNumber)-int(fecPkt.packet.SequenceNumber)) > 0x3fff { + toRemove++ + } else { + // no need to keep iterating, since receivedFECPackets is sorted + break + } + } + if toRemove > 0 { + d.receivedFECPackets = d.receivedFECPackets[toRemove:] + } + } + + switch receivedPkt.SSRC { + case d.fecSSRC: + d.insertFECPacket(receivedPkt) + case d.protectedSSRC: + d.insertMediaPacket(receivedPkt) + } + + d.discardOldRecoveredPackets() +} + +func (d *Decoder) insertMediaPacket(receivedPkt *rtp.Packet) { + for _, recoveredPacket := range d.recoveredPackets { + if recoveredPacket.SequenceNumber == receivedPkt.SequenceNumber { + return + } + } + + d.recoveredPackets = append(d.recoveredPackets, receivedPkt) + sort.Slice(d.recoveredPackets, func(i, j int) bool { + return isNewerSeq(d.recoveredPackets[i].SequenceNumber, d.recoveredPackets[j].SequenceNumber) + }) + d.updateCoveringFecPackets(receivedPkt) +} + +func (d *Decoder) updateCoveringFecPackets(receivedPkt *rtp.Packet) { + for i := range d.receivedFECPackets { + for _, pp := range d.receivedFECPackets[i].protectedPackets { + if pp.seq == receivedPkt.SequenceNumber { + pp.packet = receivedPkt + } + } + } +} + +func (d *Decoder) insertFECPacket(fecPkt *rtp.Packet) { + for i := range d.receivedFECPackets { + if d.receivedFECPackets[i].packet.SequenceNumber == fecPkt.SequenceNumber { + d.stats.FECPacketsDiscarded++ + return + } + } + + fec, err := parseFlexFEC03Header(fecPkt.Payload) + if err != nil { + d.stats.FECPacketsDiscarded++ + d.logger.Debugw("flexfec: failed to parse header", "error", err) + return + } + + if fec.protectedSSRC != d.protectedSSRC { + d.stats.FECPacketsDiscarded++ + d.logger.Debugw( + "flexfec: discarding packet protecting foreign ssrc", + "error", errUnknownProtectedSSRC, + "expectedSSRC", d.protectedSSRC, + "protectedSSRC", fec.protectedSSRC, + ) + return + } + + protectedSeqs := decodeMask(uint64(fec.mask0), 15, fec.seqNumBase) + if fec.mask1 != 0 { + protectedSeqs = append(protectedSeqs, decodeMask(uint64(fec.mask1), 31, fec.seqNumBase+15)...) + } + if fec.mask2 != 0 { + protectedSeqs = append(protectedSeqs, decodeMask(fec.mask2, 63, fec.seqNumBase+46)...) + } + + if len(protectedSeqs) == 0 { + d.stats.FECPacketsDiscarded++ + d.logger.Debugw("flexfec: discarding packet", "error", errEmptyMask) + return + } + + protectedPackets := make([]*protectedPacket, 0, len(protectedSeqs)) + protectedSeqIt := 0 + recoveredPacketIt := 0 + + for protectedSeqIt < len(protectedSeqs) && recoveredPacketIt < len(d.recoveredPackets) { + switch { + case isNewerSeq(protectedSeqs[protectedSeqIt], d.recoveredPackets[recoveredPacketIt].SequenceNumber): + protectedPackets = append(protectedPackets, &protectedPacket{ + seq: protectedSeqs[protectedSeqIt], + packet: nil, + }) + protectedSeqIt++ + case isNewerSeq(d.recoveredPackets[recoveredPacketIt].SequenceNumber, protectedSeqs[protectedSeqIt]): + recoveredPacketIt++ + default: + protectedPackets = append(protectedPackets, &protectedPacket{ + seq: protectedSeqs[protectedSeqIt], + packet: d.recoveredPackets[recoveredPacketIt], + }) + protectedSeqIt++ + recoveredPacketIt++ + } + } + + for protectedSeqIt < len(protectedSeqs) { + protectedPackets = append(protectedPackets, &protectedPacket{ + seq: protectedSeqs[protectedSeqIt], + packet: nil, + }) + protectedSeqIt++ + } + d.receivedFECPackets = append(d.receivedFECPackets, fecPacketState{ + packet: fecPkt, + flexFec: fec, + protectedPackets: protectedPackets, + }) + + sort.Slice(d.receivedFECPackets, func(i, j int) bool { + return isNewerSeq(d.receivedFECPackets[i].packet.SequenceNumber, d.receivedFECPackets[j].packet.SequenceNumber) + }) + + if len(d.receivedFECPackets) > maxFECPackets { + d.receivedFECPackets = d.receivedFECPackets[1:] + } +} + +func (d *Decoder) attemptRecovery() []*rtp.Packet { + var recoveredPackets []*rtp.Packet + for { + packetsRecovered := 0 + for i := range d.receivedFECPackets { + fecPkt := &d.receivedFECPackets[i] + packetsMissing := 0 + for _, pkt := range fecPkt.protectedPackets { + if pkt.packet == nil { + packetsMissing++ + if packetsMissing > 1 { + break + } + } + } + + if packetsMissing != 1 { + continue + } + + recovered, err := d.recoverPacket(fecPkt) + if err != nil { + d.logger.Debugw("flexfec: failed to recover packet", "error", err) + continue + } + + recoveredPackets = append(recoveredPackets, recovered) + d.recoveredPackets = append(d.recoveredPackets, recovered) + sort.Slice(d.recoveredPackets, func(i, j int) bool { + return isNewerSeq(d.recoveredPackets[i].SequenceNumber, d.recoveredPackets[j].SequenceNumber) + }) + + d.updateCoveringFecPackets(recovered) + d.discardOldRecoveredPackets() + packetsRecovered++ + } + + if packetsRecovered == 0 { + break + } + } + + return recoveredPackets +} + +func (d *Decoder) recoverPacket(fec *fecPacketState) (*rtp.Packet, error) { + // https://datatracker.ietf.org/doc/html/draft-ietf-payload-flexible-fec-scheme-03#section-6.3.2 + + // 2. For the repair packet in T, extract the FEC bit string as the + // first 80 bits of the FEC header. + headerRecovery := make([]byte, 12) + copy(headerRecovery, fec.packet.Payload[:10]) + + var seqnum uint16 + for _, pp := range fec.protectedPackets { + if pp.packet != nil { + // 1. For each of the source packets that are successfully received in + // T, compute the 80-bit string by concatenating the first 64 bits + // of their RTP header and the unsigned network-ordered 16-bit + // representation of their length in bytes minus 12. + receivedHeader, err := pp.packet.Header.Marshal() + if err != nil { + return nil, fmt.Errorf("marshal received header: %w", err) + } + binary.BigEndian.PutUint16(receivedHeader[2:4], uint16(pp.packet.MarshalSize()-12)) + for i := 0; i < 8; i++ { + headerRecovery[i] ^= receivedHeader[i] + } + } else { + seqnum = pp.seq + } + } + + // set version to 2 + headerRecovery[0] |= 0x80 + headerRecovery[0] &= 0xbf + payloadLength := binary.BigEndian.Uint16(headerRecovery[2:4]) + binary.BigEndian.PutUint16(headerRecovery[2:4], seqnum) + binary.BigEndian.PutUint32(headerRecovery[8:12], d.protectedSSRC) + + payloadRecovery := make([]byte, payloadLength) + copy(payloadRecovery, fec.flexFec.payload) + for _, pp := range fec.protectedPackets { + if pp.packet != nil { + packet, err := pp.packet.Marshal() + if err != nil { + return nil, fmt.Errorf("marshal protected packet: %w", err) + } + for i := 0; i < min(int(payloadLength), len(packet)-12); i++ { + payloadRecovery[i] ^= packet[12+i] + } + } + } + + headerRecovery = append(headerRecovery, payloadRecovery...) + + packet := &rtp.Packet{} + if err := packet.Unmarshal(headerRecovery); err != nil { + return nil, fmt.Errorf("unmarshal recovered: %w", err) + } + + return packet, nil +} + +func (d *Decoder) discardOldRecoveredPackets() { + if len(d.recoveredPackets) > recoveredPacketsLimit { + d.recoveredPackets = d.recoveredPackets[len(d.recoveredPackets)-recoveredPacketsLimit:] + } +} + +func decodeMask(mask uint64, bitCount uint16, seqNumBase uint16) []uint16 { + res := make([]uint16, 0) + for i := uint16(0); i < bitCount; i++ { + if (mask>>(bitCount-1-i))&1 == 1 { + res = append(res, seqNumBase+i) + } + } + + return res +} + +type fecPacketState struct { + packet *rtp.Packet + flexFec flexFec + protectedPackets []*protectedPacket +} + +type flexFec struct { + protectedSSRC uint32 + seqNumBase uint16 + mask0 uint16 + mask1 uint32 + mask2 uint64 + payload []byte +} + +type protectedPacket struct { + seq uint16 + packet *rtp.Packet +} + +func parseFlexFEC03Header(data []byte) (flexFec, error) { + if len(data) < 20 { + return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data)) + } + + rBit := (data[0] & 0x80) != 0 + if rBit { + return flexFec{}, errRetransmissionBitSet + } + + fBit := (data[0] & 0x40) != 0 + if fBit { + return flexFec{}, errInflexibleGeneratorMatrix + } + + ssrcCount := data[8] + if ssrcCount != 1 { + return flexFec{}, fmt.Errorf("%w: count %d", errMultipleSSRCProtection, ssrcCount) + } + + protectedSSRC := binary.BigEndian.Uint32(data[12:]) + seqNumBase := binary.BigEndian.Uint16(data[16:]) + rawPacketMask := data[18:] + var payload []byte + + kBit0 := (rawPacketMask[0] & 0x80) != 0 + maskPart0 := binary.BigEndian.Uint16(rawPacketMask[0:2]) & 0x7FFF + var maskPart1 uint32 + var maskPart2 uint64 + + if kBit0 { + payload = rawPacketMask[2:] + } else { + if len(data) < 24 { + return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data)) + } + + kBit1 := (rawPacketMask[2] & 0x80) != 0 + maskPart1 = binary.BigEndian.Uint32(rawPacketMask[2:]) & 0x7FFFFFFF + + if kBit1 { + payload = rawPacketMask[6:] + } else { + if len(data) < 32 { + return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data)) + } + + kBit2 := (rawPacketMask[6] & 0x80) != 0 + maskPart2 = binary.BigEndian.Uint64(rawPacketMask[6:]) & 0x7FFFFFFFFFFFFFFF + + if kBit2 { + payload = rawPacketMask[14:] + } else { + return flexFec{}, errLastOptionalMaskKBitSetToFalse + } + } + } + + return flexFec{ + protectedSSRC: protectedSSRC, + seqNumBase: seqNumBase, + mask0: maskPart0, + mask1: maskPart1, + mask2: maskPart2, + payload: payload, + }, nil +} + +func clonePacket(pkt *rtp.Packet) *rtp.Packet { + cloned := pkt.Clone() + return cloned +} + +func seqDiff(a, b uint16) uint16 { + return min(a-b, b-a) +} + +func absInt(x int) int { + if x >= 0 { + return x + } + + return -x +} + +func isNewerSeq(prevValue, value uint16) bool { + // half-way mark + breakpoint := uint16(0x8000) + if value-prevValue == breakpoint { + return value > prevValue + } + + return value != prevValue && (value-prevValue) < breakpoint +} diff --git a/pkg/sfu/flexfec/decoder_test.go b/pkg/sfu/flexfec/decoder_test.go new file mode 100644 index 000000000..3431c099e --- /dev/null +++ b/pkg/sfu/flexfec/decoder_test.go @@ -0,0 +1,355 @@ +// 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 ( + "encoding/binary" + "math/rand" + "testing" + + pionflexfec "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/logger" +) + +const ( + testFECSSRC = uint32(1234) + testMediaSSRC = uint32(5678) + testFECPT = uint8(115) + testMediaPT = uint8(96) +) + +func makeMediaPackets(t *testing.T, baseSN uint16, count int) []rtp.Packet { + t.Helper() + rng := rand.New(rand.NewSource(int64(baseSN))) + packets := make([]rtp.Packet, 0, count) + for i := 0; i < count; i++ { + payload := make([]byte, 100+rng.Intn(900)) + rng.Read(payload) + packets = append(packets, rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: testMediaPT, + SequenceNumber: baseSN + uint16(i), + Timestamp: 3000 * uint32(i), + SSRC: testMediaSSRC, + Marker: i == count-1, + }, + Payload: payload, + }) + } + return packets +} + +func encodeFEC(t *testing.T, mediaPackets []rtp.Packet, numFEC uint32) []rtp.Packet { + t.Helper() + encoder := pionflexfec.NewFlexEncoder03(testFECPT, testFECSSRC) + fecPackets := encoder.EncodeFec(mediaPackets, numFEC) + require.NotEmpty(t, fecPackets) + return fecPackets +} + +func requirePacketEqual(t *testing.T, expected *rtp.Packet, actual *rtp.Packet) { + t.Helper() + require.Equal(t, expected.SequenceNumber, actual.SequenceNumber) + require.Equal(t, expected.Timestamp, actual.Timestamp) + require.Equal(t, expected.PayloadType, actual.PayloadType) + require.Equal(t, expected.SSRC, actual.SSRC) + require.Equal(t, expected.Marker, actual.Marker) + require.Equal(t, expected.Payload, actual.Payload) +} + +func TestDecoderRecoversSingleLoss(t *testing.T) { + media := makeMediaPackets(t, 100, 5) + fec := encodeFEC(t, media, 1) + + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + + // drop media[2], feed the rest + var recovered []*rtp.Packet + for i := range media { + if i == 2 { + continue + } + recovered = append(recovered, decoder.DecodeFec(&media[i])...) + } + require.Empty(t, recovered) + + for i := range fec { + recovered = append(recovered, decoder.DecodeFec(&fec[i])...) + } + + require.Len(t, recovered, 1) + requirePacketEqual(t, &media[2], recovered[0]) + + stats := decoder.Stats() + assert.Equal(t, uint64(len(fec)), stats.FECPacketsReceived) + assert.Equal(t, uint64(1), stats.PacketsRecovered) + assert.Equal(t, uint64(0), stats.FECPacketsDiscarded) +} + +func TestDecoderRecoversWithLateMedia(t *testing.T) { + // FEC arrives while two packets are missing; recovery happens once one of + // them shows up late. Exercises updateCoveringFecPackets and the + // stable-pointer window. + media := makeMediaPackets(t, 200, 5) + fec := encodeFEC(t, media, 1) + + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + + var recovered []*rtp.Packet + for _, i := range []int{0, 3, 4} { + recovered = append(recovered, decoder.DecodeFec(&media[i])...) + } + for i := range fec { + recovered = append(recovered, decoder.DecodeFec(&fec[i])...) + } + // two packets missing from the protected window, nothing recoverable yet + require.Empty(t, recovered) + + // late arrival of media[1] leaves only media[2] missing + recovered = decoder.DecodeFec(&media[1]) + require.Len(t, recovered, 1) + requirePacketEqual(t, &media[2], recovered[0]) +} + +func TestDecoderRecoversMultipleWindows(t *testing.T) { + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + encoder := pionflexfec.NewFlexEncoder03(testFECPT, testFECSSRC) + + var allRecovered []*rtp.Packet + dropped := make(map[uint16]*rtp.Packet) + baseSN := uint16(1000) + for window := 0; window < 10; window++ { + media := makeMediaPackets(t, baseSN, 10) + fecPackets := encoder.EncodeFec(media, 1) + require.NotEmpty(t, fecPackets) + + dropIdx := window % 10 + for i := range media { + if i == dropIdx { + dropped[media[i].SequenceNumber] = &media[i] + continue + } + allRecovered = append(allRecovered, decoder.DecodeFec(&media[i])...) + } + for i := range fecPackets { + allRecovered = append(allRecovered, decoder.DecodeFec(&fecPackets[i])...) + } + baseSN += 10 + } + + require.Len(t, allRecovered, 10) + for _, rec := range allRecovered { + expected, ok := dropped[rec.SequenceNumber] + require.True(t, ok, "recovered unexpected sequence number %d", rec.SequenceNumber) + requirePacketEqual(t, expected, rec) + } + assert.Equal(t, uint64(10), decoder.Stats().PacketsRecovered) +} + +func TestDecoderSequenceNumberWrap(t *testing.T) { + media := makeMediaPackets(t, 65533, 5) // spans 65533..1 + fec := encodeFEC(t, media, 1) + + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + + var recovered []*rtp.Packet + for i := range media { + if i == 3 { // sequence number 0 + continue + } + recovered = append(recovered, decoder.DecodeFec(&media[i])...) + } + for i := range fec { + recovered = append(recovered, decoder.DecodeFec(&fec[i])...) + } + + require.Len(t, recovered, 1) + requirePacketEqual(t, &media[3], recovered[0]) +} + +func TestDecoderDiscardsForeignProtectedSSRC(t *testing.T) { + media := makeMediaPackets(t, 300, 5) + fec := encodeFEC(t, media, 1) + + // decoder bound to a different protected stream + decoder := NewDecoder(testFECSSRC, testMediaSSRC+1, logger.GetLogger()) + recovered := decoder.DecodeFec(&fec[0]) + require.Empty(t, recovered) + + stats := decoder.Stats() + assert.Equal(t, uint64(1), stats.FECPacketsReceived) + assert.Equal(t, uint64(1), stats.FECPacketsDiscarded) +} + +func TestDecoderDiscardsMalformedFEC(t *testing.T) { + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + + for _, payload := range [][]byte{ + nil, + {0x00}, + make([]byte, 10), + func() []byte { // R bit set + p := make([]byte, pionflexfec.BaseFec03HeaderSize+4) + p[0] = 0x80 + p[8] = 1 + return p + }(), + func() []byte { // multiple protected ssrcs + p := make([]byte, pionflexfec.BaseFec03HeaderSize+4) + p[8] = 2 + return p + }(), + func() []byte { // empty mask + p := make([]byte, pionflexfec.BaseFec03HeaderSize+4) + p[8] = 1 + binary.BigEndian.PutUint32(p[12:], testMediaSSRC) + binary.BigEndian.PutUint16(p[18:], 0x8000) + return p + }(), + } { + pkt := &rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: testFECPT, + SequenceNumber: uint16(rand.Intn(65536)), + SSRC: testFECSSRC, + }, + Payload: payload, + } + require.NotPanics(t, func() { + require.Empty(t, decoder.DecodeFec(pkt)) + }) + } + + stats := decoder.Stats() + assert.Equal(t, uint64(6), stats.FECPacketsReceived) + assert.Equal(t, uint64(6), stats.FECPacketsDiscarded) +} + +func TestDecoderDiscardsDuplicateFEC(t *testing.T) { + media := makeMediaPackets(t, 400, 5) + fec := encodeFEC(t, media, 1) + + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + for i := range media { + decoder.DecodeFec(&media[i]) + } + require.Empty(t, decoder.DecodeFec(&fec[0])) + require.Empty(t, decoder.DecodeFec(&fec[0])) + + stats := decoder.Stats() + assert.Equal(t, uint64(2), stats.FECPacketsReceived) + assert.Equal(t, uint64(1), stats.FECPacketsDiscarded) +} + +func TestDecoderInputMemoryReuse(t *testing.T) { + // the decoder must not retain references to caller-owned packet memory + media := makeMediaPackets(t, 500, 5) + fec := encodeFEC(t, media, 1) + + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + + scratch := &rtp.Packet{} + feed := func(src *rtp.Packet) []*rtp.Packet { + buf, err := src.Marshal() + require.NoError(t, err) + require.NoError(t, scratch.Unmarshal(buf)) + out := decoder.DecodeFec(scratch) + // clobber the scratch memory the decoder saw + for i := range scratch.Payload { + scratch.Payload[i] = 0xde + } + return out + } + + var recovered []*rtp.Packet + for i := range media { + if i == 2 { + continue + } + recovered = append(recovered, feed(&media[i])...) + } + for i := range fec { + recovered = append(recovered, feed(&fec[i])...) + } + + require.Len(t, recovered, 1) + requirePacketEqual(t, &media[2], recovered[0]) +} + +func TestDecoderTwoFECPacketsTwoLosses(t *testing.T) { + // with 2 FEC packets over 10 media packets, the coverage interleaves, so + // two losses landing in different coverage groups are both recoverable + media := makeMediaPackets(t, 600, 10) + fec := encodeFEC(t, media, 2) + require.Len(t, fec, 2) + + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + + var recovered []*rtp.Packet + for i := range media { + if i == 2 || i == 3 { + continue + } + recovered = append(recovered, decoder.DecodeFec(&media[i])...) + } + for i := range fec { + recovered = append(recovered, decoder.DecodeFec(&fec[i])...) + } + + recoveredSNs := make(map[uint16]bool) + for _, r := range recovered { + recoveredSNs[r.SequenceNumber] = true + } + // at least one of the two losses must be recovered; both when the losses + // fall in distinct coverage groups + require.NotEmpty(t, recovered) + for _, r := range recovered { + expectedIdx := int(r.SequenceNumber - 600) + requirePacketEqual(t, &media[expectedIdx], r) + } + require.True(t, recoveredSNs[602] || recoveredSNs[603]) +} + +func TestDecoderResetsOnBigSequenceGap(t *testing.T) { + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + + media := makeMediaPackets(t, 100, 110) + for i := range media { + decoder.DecodeFec(&media[i]) + } + + // jump far ahead, decoder should reset its windows rather than misuse + // stale state + farMedia := makeMediaPackets(t, 30000, 5) + fec := encodeFEC(t, farMedia, 1) + var recovered []*rtp.Packet + for i := range farMedia { + if i == 1 { + continue + } + recovered = append(recovered, decoder.DecodeFec(&farMedia[i])...) + } + for i := range fec { + recovered = append(recovered, decoder.DecodeFec(&fec[i])...) + } + require.Len(t, recovered, 1) + requirePacketEqual(t, &farMedia[1], recovered[0]) +} diff --git a/pkg/sfu/pacer/base.go b/pkg/sfu/pacer/base.go index 96e3be250..afcdcec41 100644 --- a/pkg/sfu/pacer/base.go +++ b/pkg/sfu/pacer/base.go @@ -76,8 +76,7 @@ func (b *Base) SendPacket(p *Packet) (int, error) { return 0, err } - var written int - written, err = p.WriteStream.WriteRTP(p.Header, p.Payload) + written, err := p.WriteStream.WriteRTP(p.Header, p.Payload) if err != nil { if !errors.Is(err, io.ErrClosedPipe) { b.logger.Errorw("write rtp packet failed", err) @@ -85,6 +84,10 @@ func (b *Base) SendPacket(p *Packet) (int, error) { return 0, err } + if p.OnSent != nil { + p.OnSent(p.Header, p.Payload) + } + return written, nil } diff --git a/pkg/sfu/pacer/pacer.go b/pkg/sfu/pacer/pacer.go index f23212519..45536175b 100644 --- a/pkg/sfu/pacer/pacer.go +++ b/pkg/sfu/pacer/pacer.go @@ -54,6 +54,11 @@ type Packet struct { WriteStream webrtc.TrackLocalWriter Pool *sync.Pool PoolEntity *[]byte + // OnSent is invoked after the packet has been written to the wire with + // the final header (pacer patched extensions included). Header and + // payload are pooled after the call returns, implementations must copy + // anything they keep. + OnSent func(hdr *rtp.Header, payload []byte) } type Pacer interface { diff --git a/pkg/sfu/utils/helpers.go b/pkg/sfu/utils/helpers.go index 7ce7eaac8..b969f4099 100644 --- a/pkg/sfu/utils/helpers.go +++ b/pkg/sfu/utils/helpers.go @@ -17,6 +17,7 @@ package utils import ( "errors" "fmt" + "strings" "github.com/pion/interceptor" "github.com/pion/rtp" @@ -59,6 +60,18 @@ func FindRTXPayloadType(needle webrtc.PayloadType, haystack []webrtc.RTPCodecPar return webrtc.PayloadType(0) } +// FindFlexFECPayloadType returns the negotiated flexfec-03 payload type, or 0 +// if flexfec was not negotiated. +func FindFlexFECPayloadType(haystack []webrtc.RTPCodecParameters) webrtc.PayloadType { + for _, c := range haystack { + if strings.EqualFold(c.MimeType, webrtc.MimeTypeFlexFEC03) { + return c.PayloadType + } + } + + return webrtc.PayloadType(0) +} + // GetHeaderExtensionID returns the ID of a header extension, or 0 if not found func GetHeaderExtensionID(extensions []interceptor.RTPHeaderExtension, extension webrtc.RTPHeaderExtensionCapability) int { for _, h := range extensions { diff --git a/pkg/telemetry/prometheus/packets.go b/pkg/telemetry/prometheus/packets.go index a9f6684e1..6306b3e98 100644 --- a/pkg/telemetry/prometheus/packets.go +++ b/pkg/telemetry/prometheus/packets.go @@ -56,6 +56,9 @@ var ( promPacketLabels = []string{"direction", "transmission", "country"} promPacketTotal *prometheus.CounterVec promPacketBytes *prometheus.CounterVec + promFECLabels = []string{"direction", "type"} + promFECPacketTotal *prometheus.CounterVec + promFECPacketBytes *prometheus.CounterVec promRTCPLabels = []string{"direction", "country"} promStreamLabels = []string{"direction", "source", "type", "country"} promNackTotal *prometheus.CounterVec @@ -87,6 +90,18 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) { Name: "bytes", ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, }, promPacketLabels) + promFECPacketTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: livekitNamespace, + Subsystem: "fec", + Name: "packets", + ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, + }, promFECLabels) + promFECPacketBytes = prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: livekitNamespace, + Subsystem: "fec", + Name: "bytes", + ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, + }, promFECLabels) promNackTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: livekitNamespace, Subsystem: "nack", @@ -192,6 +207,8 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) { prometheus.MustRegister(promPacketTotal) prometheus.MustRegister(promPacketBytes) + prometheus.MustRegister(promFECPacketTotal) + prometheus.MustRegister(promFECPacketBytes) prometheus.MustRegister(promNackTotal) prometheus.MustRegister(promPliTotal) prometheus.MustRegister(promFirTotal) @@ -208,6 +225,35 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) { prometheus.MustRegister(promForwardLatencyHist) } +// RecordFECUpstream tracks FlexFEC activity on the publisher leg: FEC +// packets received from publishers, media packets recovered with them and +// FEC packets that could not be used. +func RecordFECUpstream(received int, recovered int, discarded int, bytesReceived uint64) { + if received > 0 { + promFECPacketTotal.WithLabelValues(string(Incoming), "received").Add(float64(received)) + } + if recovered > 0 { + promFECPacketTotal.WithLabelValues(string(Incoming), "recovered").Add(float64(recovered)) + } + if discarded > 0 { + promFECPacketTotal.WithLabelValues(string(Incoming), "discarded").Add(float64(discarded)) + } + if bytesReceived > 0 { + promFECPacketBytes.WithLabelValues(string(Incoming), "received").Add(float64(bytesReceived)) + } +} + +// RecordFECDownstreamSent tracks FlexFEC packets generated toward +// subscribers. +func RecordFECDownstreamSent(packets int, bytes uint64) { + if packets > 0 { + promFECPacketTotal.WithLabelValues(string(Outgoing), "sent").Add(float64(packets)) + } + if bytes > 0 { + promFECPacketBytes.WithLabelValues(string(Outgoing), "sent").Add(float64(bytes)) + } +} + func IncrementPackets(country string, direction Direction, count uint64, retransmit bool) { var transmission TransmissionType if retransmit { diff --git a/test/client/client.go b/test/client/client.go index bb38cdaf3..b4286859a 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -65,6 +65,7 @@ type RTCClient struct { subscriber *rtc.PCTransport enabledCodecs []*livekit.Codec forceRelay bool + enableFlexFEC bool transportReady chan struct{} // sid => track localTracks map[string]webrtc.TrackLocal @@ -141,6 +142,7 @@ type Options struct { UseJoinRequestQueryParam bool RTCServicePath string ForceRelay bool + EnableFlexFEC bool } func NewWebSocketConn(host, token string, opts *Options) (*websocket.Conn, error) { @@ -276,6 +278,7 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti c.signalRequestInterceptor = opts.SignalRequestInterceptor c.signalResponseInterceptor = opts.SignalResponseInterceptor c.forceRelay = opts.ForceRelay + c.enableFlexFEC = opts.EnableFlexFEC } return c, nil @@ -294,6 +297,17 @@ func (c *RTCClient) createTransport(rtcconf webrtc.Configuration) error { ff := buffer.NewFactoryOfBufferFactory(500, 200) conf.SetBufferFactory(ff.CreateBufferFactory()) + if c.enableFlexFEC { + fecConfig := rtc.FlexFECDirectionConfig{ + Enabled: true, + PayloadType: 115, + NumMediaPackets: 5, + NumFECPackets: 2, + } + conf.Publisher.FlexFEC = fecConfig + conf.Subscriber.FlexFEC = fecConfig + } + // // The signal targets are from point of view of server. // From client side, they are flipped, @@ -1106,6 +1120,11 @@ func (c *RTCClient) LastAnswer() *webrtc.SessionDescription { return c.publisher.CurrentRemoteDescription() } +// LastSubscriberOffer returns the SDP of the last offer received on the subscriber connection +func (c *RTCClient) LastSubscriberOffer() *webrtc.SessionDescription { + return c.subscriber.CurrentRemoteDescription() +} + func (c *RTCClient) ensurePublisherConnected() error { select { case <-c.transportReady: diff --git a/test/flexfec_test.go b/test/flexfec_test.go new file mode 100644 index 000000000..fdb718784 --- /dev/null +++ b/test/flexfec_test.go @@ -0,0 +1,178 @@ +// 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 test + +import ( + "fmt" + "strings" + "testing" + + prom "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/config" + "github.com/livekit/livekit-server/pkg/service" + "github.com/livekit/livekit-server/pkg/testutils" + testclient "github.com/livekit/livekit-server/test/client" + "github.com/livekit/protocol/logger" +) + +func fecCounterValue(t *testing.T, direction string, typ string) float64 { + t.Helper() + families, err := prom.DefaultGatherer.Gather() + require.NoError(t, err) + for _, family := range families { + if family.GetName() != "livekit_fec_packets" { + continue + } + for _, metric := range family.GetMetric() { + matchedDirection, matchedType := false, false + for _, label := range metric.GetLabel() { + if label.GetName() == "direction" && label.GetValue() == direction { + matchedDirection = true + } + if label.GetName() == "type" && label.GetValue() == typ { + matchedType = true + } + } + if matchedDirection && matchedType { + return metric.GetCounter().GetValue() + } + } + } + return 0 +} + +func setupFlexFECTest(name string) (*service.LivekitServer, func()) { + logger.Infow("----------------STARTING TEST----------------", "test", name) + s := createSingleNodeServer(func(conf *config.Config) { + conf.RTC.FlexFEC = config.FlexFECConfig{ + UpstreamEnabled: true, + DownstreamEnabled: true, + PayloadType: 115, + NumMediaPackets: 5, + NumFECPackets: 2, + } + }) + go func() { + if err := s.Start(); err != nil { + logger.Errorw("server returned error", err) + } + }() + + waitForServerToStart(s) + + return s, func() { + s.Stop(true) + logger.Infow("----------------FINISHING TEST----------------", "test", name) + } +} + +// TestFlexFEC verifies flexfec-03 negotiation on both legs and that the SFU +// generates FEC toward subscribers: a publisher with FlexFEC enabled offers +// flexfec + FEC-FR which the SFU accepts (upstream), and the SFU's offer to +// a FlexFEC capable subscriber carries a FEC-FR repair stream that the +// DownTrack populates with repair packets (downstream). +func TestFlexFEC(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, finish := setupFlexFECTest("TestFlexFEC") + defer finish() + + fecSentBefore := fecCounterValue(t, "outgoing", "sent") + + opts := &testclient.Options{AutoSubscribe: true, EnableFlexFEC: true} + c1 := createRTCClient("fec_pub", defaultServerPort, testRTCServicePathv0, opts) + c2 := createRTCClient("fec_sub", defaultServerPort, testRTCServicePathv0, opts) + defer stopClients(c1, c2) + waitUntilConnected(t, c1, c2) + + writer, err := c1.AddStaticTrack("video/vp8", "video", "fecvideo") + require.NoError(t, err) + defer writer.Stop() + + // publisher leg: the SFU's answer must accept the offered flexfec-03 + testutils.WithTimeout(t, func() string { + sd := c1.LastAnswer() + if sd == nil { + return "no answer received on publisher connection" + } + if !strings.Contains(sd.SDP, "flexfec-03") { + return "SFU answer does not contain flexfec-03" + } + return "" + }) + + // subscriber leg: the SFU's offer must announce the FEC repair stream + // and the DownTrack must generate FEC packets for the forwarded media + testutils.WithTimeout(t, func() string { + tracks := c2.SubscribedTracks() + if len(tracks[c1.ID()]) == 0 { + return "c2 was not subscribed to c1's tracks" + } + + sd := c2.LastSubscriberOffer() + if sd == nil { + return "no offer received on subscriber connection" + } + if !strings.Contains(sd.SDP, "flexfec-03") { + return "SFU offer does not contain flexfec-03" + } + if !strings.Contains(sd.SDP, "FEC-FR") { + return "SFU offer does not contain a FEC-FR ssrc-group" + } + + fecSent := fecCounterValue(t, "outgoing", "sent") + if fecSent <= fecSentBefore { + return fmt.Sprintf("no FEC packets generated, counter at %f", fecSent) + } + return "" + }) +} + +// TestFlexFECDisabledClient ensures media still flows when the server has +// FlexFEC enabled but a client does not negotiate it. +func TestFlexFECDisabledClient(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, finish := setupFlexFECTest("TestFlexFECDisabledClient") + defer finish() + + c1 := createRTCClient("nofec_pub", defaultServerPort, testRTCServicePathv0, nil) + c2 := createRTCClient("nofec_sub", defaultServerPort, testRTCServicePathv0, nil) + defer stopClients(c1, c2) + waitUntilConnected(t, c1, c2) + + writer, err := c1.AddStaticTrack("video/vp8", "video", "plainvideo") + require.NoError(t, err) + defer writer.Stop() + + testutils.WithTimeout(t, func() string { + tracks := c2.SubscribedTracks() + if len(tracks[c1.ID()]) == 0 { + return "c2 was not subscribed to c1's tracks" + } + if c2.BytesReceived() == 0 { + return "c2 did not receive any media" + } + return "" + }) +}