Add upstream FlexFEC recovery support

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,

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{

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,

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()) {

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()

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 <media ssrc> <fec ssrc>` (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 {

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
+	}
+}

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) {

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:

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")
+}

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)
+	}
+}

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
+}

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])
+}

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 {

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,

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 ""
+	})
+}