From 6a879f4f48e841ab4d9fa6c431c8cb6442e82502 Mon Sep 17 00:00:00 2001 From: David Chen Date: Fri, 28 Aug 2026 17:46:41 -0700 Subject: [PATCH] Add upstream FlexFEC recovery support --- config-sample.yaml | 7 + pkg/config/config.go | 25 ++ pkg/rtc/config.go | 28 +- pkg/rtc/mediaengine.go | 68 ++++ pkg/rtc/mediatrack.go | 5 + pkg/rtc/transport.go | 63 ++++ pkg/rtc/transport_fec_test.go | 125 +++++++ 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/flexfec/decoder.go | 517 ++++++++++++++++++++++++++++ pkg/sfu/flexfec/decoder_test.go | 355 +++++++++++++++++++ pkg/telemetry/prometheus/packets.go | 35 ++ test/client/client.go | 12 + test/flexfec_upstream_test.go | 114 ++++++ 16 files changed, 1882 insertions(+), 3 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/flexfec/decoder.go create mode 100644 pkg/sfu/flexfec/decoder_test.go create mode 100644 test/flexfec_upstream_test.go diff --git a/config-sample.yaml b/config-sample.yaml index 5c1c50a42..73ab0115e 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -117,6 +117,13 @@ 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, disabled by default. + # # Accepts FEC from publishers and uses it to repair lost packets before + # # forwarding. + # flexfec: + # upstream_enabled: false + # # must not collide with codec or RTX payload types + # payload_type: 115 # # 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 2ef6a3efe..32936d32f 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -142,6 +142,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"` @@ -220,6 +222,28 @@ type CongestionControlConfig struct { SendSideBWE sendsidebwe.SendSideBWEConfig `yaml:"send_side_bwe,omitempty"` } +// FlexFECConfig controls FlexFEC-03 recovery on the publisher -> SFU leg. +type FlexFECConfig struct { + // negotiate flexfec-03 with publishers and use it to recover lost upstream packets + UpstreamEnabled bool `yaml:"upstream_enabled,omitempty"` + // payload type used for flexfec-03, must not collide with codec payload + // types or their RTX (pt+1) slots + PayloadType uint8 `yaml:"payload_type,omitempty"` +} + +var DefaultFlexFECConfig = FlexFECConfig{ + UpstreamEnabled: false, + PayloadType: 115, +} + +// WithDefaults returns a copy with zero values replaced by defaults. +func (c FlexFECConfig) WithDefaults() FlexFECConfig { + if c.PayloadType == 0 { + c.PayloadType = DefaultFlexFECConfig.PayloadType + } + return c +} + type PlayoutDelayConfig struct { Enabled bool `yaml:"enabled,omitempty"` Min int `yaml:"min,omitempty"` @@ -525,6 +549,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..37ea20330 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,13 @@ type RTCPFeedbackConfig struct { type DirectionConfig struct { RTPHeaderExtension RTPHeaderExtensionConfig RTCPFeedback RTCPFeedbackConfig + FlexFEC FlexFECDirectionConfig +} + +// FlexFECDirectionConfig enables flexfec-03 for a transport direction. +type FlexFECDirectionConfig struct { + Enabled bool + PayloadType uint8 } func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { @@ -80,19 +89,26 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { rtcConf.PacketBufferSizeAudio = rtcConf.PacketBufferSize } + flexFEC := rtcConf.FlexFEC.WithDefaults() + if flexFEC.UpstreamEnabled { + 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), + Publisher: getPublisherConfig(false, flexFEC), Subscriber: getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE), + 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) { @@ -104,9 +120,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 +162,7 @@ func getPublisherConfig(consolidated bool) DirectionConfig { } return DirectionConfig{ + FlexFEC: publisherFlexFEC, RTPHeaderExtension: RTPHeaderExtensionConfig{ Audio: []string{ sdp.SDESMidURI, diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index 8f44593f7..b148054c6 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -25,6 +25,55 @@ import ( "github.com/livekit/protocol/livekit" ) +// flexFECRepairWindow is the flexfec-03 "repair-window" fmtp value in +// microseconds (10 s), matching pion's ConfigureFlexFEC03 default. +const flexFECRepairWindow = 10_000_000 + +// 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: fmt.Sprintf("repair-window=%d", flexFECRepairWindow), + 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 +} + type codecToRegister struct { webrtc.RTPCodecParameters strictFmtp bool @@ -137,6 +186,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 } @@ -204,6 +262,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 { @@ -211,6 +270,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 3c167e0f2..5bc6cce9c 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" ) @@ -568,6 +569,10 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe buff.OnNotifyRTX(t.MediaTrackReceiver.setLayerRtxInfo) + buff.OnFECRecovery(func(received int, recovered 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/transport.go b/pkg/rtc/transport.go index 2f9f7d440..7a9e4ffe5 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -1024,6 +1024,7 @@ func (t *PCTransport) queueOrConfigureSender( filterOutH264HighProfile: !t.params.IsOfferer, enableAudioStereo: enableAudioStereo, enableAudioNACK: enableAudioNACK, + keepFlexFEC: t.params.DirectionConfig.FlexFEC.Enabled, } if !t.params.IsOfferer { t.sendersPendingConfigMu.Lock() @@ -1693,6 +1694,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 } @@ -2941,6 +2951,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 } @@ -3114,6 +3133,7 @@ type configureSenderParams struct { filterOutH264HighProfile bool enableAudioStereo bool enableAudioNACK bool + keepFlexFEC bool } func configureSender(params configureSenderParams, offerAudioPT map[mime.MimeType]webrtc.PayloadType) { @@ -3122,6 +3142,7 @@ func configureSender(params configureSenderParams, offerAudioPT map[mime.MimeTyp params.enabledCodecs, params.rtcpFeedbackConfig, params.filterOutH264HighProfile, + params.keepFlexFEC, ) if params.transceiver.Kind() == webrtc.RTPCodecTypeAudio { @@ -3227,6 +3248,7 @@ func configureSenderCodecs( enabledCodecs []*livekit.Codec, rtcpFeedbackConfig RTCPFeedbackConfig, filterOutH264HighProfile bool, + keepFlexFEC bool, ) { if len(enabledCodecs) == 0 { return @@ -3242,6 +3264,7 @@ func configureSenderCodecs( enabledCodecs, rtcpFeedbackConfig, filterOutH264HighProfile, + keepFlexFEC, ) tr.SetCodecPreferences(filteredCodecs) } @@ -3264,6 +3287,7 @@ func (t *PCTransport) restrictReceiverCodecsToPublishList() { t.params.EnabledPublishCodecs, t.params.DirectionConfig.RTCPFeedback, false, + t.params.DirectionConfig.FlexFEC.Enabled, ) if len(filtered) == 0 { continue @@ -3315,6 +3339,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...) @@ -3365,6 +3397,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/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 5468b3b94..462373acf 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" @@ -73,6 +74,12 @@ type Buffer struct { primaryBufferForRTX *Buffer rtxPktBuf []byte + primaryBufferForFEC *Buffer + fecSSRC uint32 + fecDecoder *flexfec.Decoder + fecPktBuf []byte + onFECRecovery func(received int, recovered int, discarded int, bytesReceived int) + streamInfoProbe *StreamInfoProbe warnedPendingOverflow bool } @@ -144,6 +151,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 { @@ -201,6 +209,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) @@ -236,6 +257,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 { @@ -358,6 +384,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 since the previous callback: FEC packets received, media packets +// recovered, FEC packets discarded and FEC bytes received. +func (b *Buffer) OnFECRecovery(fn func(received int, recovered 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(received, len(recovered), 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 bf373a8e5..4d6cb0384 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] @@ -313,6 +314,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..435b3c1e2 --- /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(received int, recovered 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 8fb6d46dd..2f3e11bf1 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 @@ -147,3 +165,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/flexfec/decoder.go b/pkg/sfu/flexfec/decoder.go new file mode 100644 index 000000000..fdac3f11d --- /dev/null +++ b/pkg/sfu/flexfec/decoder.go @@ -0,0 +1,517 @@ +// 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 +) + +// FlexFEC-03 header bit fields. +// https://datatracker.ietf.org/doc/html/draft-ietf-payload-flexible-fec-scheme-03#section-6.1 +const ( + fecRetransmissionBit = 0x80 // R bit, first FEC header byte + fecInflexibleBit = 0x40 // F bit, first FEC header byte + fecMaskKBit = 0x80 // K bit, terminates the run of packet-mask chunks + + // Data-bit width of each packet-mask chunk (the chunk minus its K bit). + fecMask0Bits = 15 + fecMask1Bits = 31 + fecMask2Bits = 63 + + // Value masks that clear the K bit from each packet-mask chunk. + fecMask0Value = 0x7FFF + fecMask1Value = 0x7FFFFFFF + fecMask2Value = 0x7FFFFFFFFFFFFFFF +) + +// 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 := receivedPacket.Clone() + + 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), fecMask0Bits, fec.seqNumBase) + if fec.mask1 != 0 { + protectedSeqs = append(protectedSeqs, decodeMask(uint64(fec.mask1), fecMask1Bits, fec.seqNumBase+fecMask0Bits)...) + } + if fec.mask2 != 0 { + protectedSeqs = append(protectedSeqs, decodeMask(fec.mask2, fecMask2Bits, fec.seqNumBase+fecMask0Bits+fecMask1Bits)...) + } + + 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] & fecRetransmissionBit) != 0 + if rBit { + return flexFec{}, errRetransmissionBitSet + } + + fBit := (data[0] & fecInflexibleBit) != 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] & fecMaskKBit) != 0 + maskPart0 := binary.BigEndian.Uint16(rawPacketMask[0:2]) & fecMask0Value + 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] & fecMaskKBit) != 0 + maskPart1 = binary.BigEndian.Uint32(rawPacketMask[2:]) & fecMask1Value + + if kBit1 { + payload = rawPacketMask[6:] + } else { + if len(data) < 32 { + return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data)) + } + + kBit2 := (rawPacketMask[6] & fecMaskKBit) != 0 + maskPart2 = binary.BigEndian.Uint64(rawPacketMask[6:]) & fecMask2Value + + 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 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/telemetry/prometheus/packets.go b/pkg/telemetry/prometheus/packets.go index 37e67e895..406b6bdbc 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: "total", + 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,24 @@ 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)) + } +} + 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 bd8c76f20..fca2afee6 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,15 @@ 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, + } + conf.Publisher.FlexFEC = fecConfig + conf.Subscriber.FlexFEC = fecConfig + } + // // The signal targets are from point of view of server. // From client side, they are flipped, diff --git a/test/flexfec_upstream_test.go b/test/flexfec_upstream_test.go new file mode 100644 index 000000000..8c80e9e39 --- /dev/null +++ b/test/flexfec_upstream_test.go @@ -0,0 +1,114 @@ +// 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 ( + "strings" + "testing" + + "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 setupFlexFECUpstreamTest(name string) (*service.LivekitServer, func()) { + logger.Infow("----------------STARTING TEST----------------", "test", name) + s := createSingleNodeServer(func(conf *config.Config) { + conf.RTC.FlexFEC = config.FlexFECConfig{ + UpstreamEnabled: true, + PayloadType: 115, + } + }) + 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) + } +} + +// TestFlexFECUpstreamNegotiation verifies that the SFU accepts flexfec-03 +// offered by a publisher when upstream recovery is enabled. +func TestFlexFECUpstreamNegotiation(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, finish := setupFlexFECUpstreamTest("TestFlexFECUpstreamNegotiation") + defer finish() + + opts := &testclient.Options{AutoSubscribe: true, EnableFlexFEC: true} + c1 := createRTCClient("fec_pub", defaultServerPort, testRTCServicePathv0, opts) + defer stopClients(c1) + waitUntilConnected(t, c1) + + writer, err := c1.AddStaticTrack("video/vp8", "video", "fecvideo") + require.NoError(t, err) + defer writer.Stop() + + 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 "" + }) +} + +// 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 := setupFlexFECUpstreamTest("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 "" + }) +}