initial fec implementation

config-sample.yaml

@@ -115,6 +115,24 @@ rtc:
   # packet_buffer_size_video: 500
   # # number of packets to buffer in the SFU for audio, defaults to 200
   # packet_buffer_size_audio: 200
+  # # FlexFEC-03 (RFC 8627) video forward error correction. Disabled by default.
+  # # FlexFEC is terminated at the SFU: it cannot be passed through unchanged because the
+  # # SFU rewrites SSRC/sequence numbers per subscriber. The two directions are independent.
+  # flexfec:
+  #   # generate a fresh FlexFEC repair stream per subscriber (downstream loss recovery)
+  #   subscriber: false
+  #   # negotiate, receive and decode a publisher's FlexFEC repair stream (uplink loss recovery)
+  #   publisher: false
+  #   # dynamic payload type used for the flexfec-03 codec, defaults to 49
+  #   payload_type: 49
+  #   # subscriber-side generation is a fixed, block-based scheme: for every
+  #   # num_media_packets media packets, num_fec_packets repair packets are emitted.
+  #   # bandwidth overhead ~= num_fec_packets/num_media_packets, and a block can recover
+  #   # at most num_fec_packets losses out of num_media_packets. smaller num_media_packets
+  #   # recovers faster (lower added latency) but costs more overhead. defaults below give
+  #   # ~20% overhead; the pion library default (no tuning) is a much heavier 5/2 = 40%.
+  #   num_media_packets: 10
+  #   num_fec_packets: 2
   # # minimum amount of time between pli/fir rtcp packets being sent to an individual
   # # producer. Increasing these times can lead to longer black screens when new participants join,
   # # while reducing them can lead to higher stream bitrate.

pkg/config/config.go

@@ -138,6 +138,67 @@ type RTCConfig struct {
 
 	// enable rtp stream restart detection for published tracks
 	EnableRTPStreamRestartDetection bool `yaml:"enable_rtp_stream_restart_detection,omitempty"`
+
+	// FlexFEC (RFC 8627 / flexfec-03) support
+	FlexFEC FlexFECConfig `yaml:"flexfec,omitempty"`
+}
+
+// FlexFECConfig controls FlexFEC-03 (video) forward error correction.
+//
+// FlexFEC cannot be passed through an SFU as-is because the SFU rewrites SSRC and
+// sequence numbers per subscriber. Instead, FlexFEC is terminated at the SFU and
+// the two directions are handled independently:
+//   - Subscriber: the SFU generates a fresh FlexFEC repair stream per downstream so a
+//     subscriber can recover packets lost on the SFU -> subscriber path.
+//   - Publisher: the SFU negotiates, receives and decodes a publisher's FlexFEC repair
+//     stream to recover packets lost on the publisher -> SFU path.
+//
+// Both are disabled by default. FlexFEC is video-only here; audio keeps RED / in-band
+// Opus FEC.
+type FlexFECConfig struct {
+	// generate FlexFEC for subscribers (downstream loss recovery)
+	Subscriber bool `yaml:"subscriber,omitempty"`
+	// negotiate, receive and decode FlexFEC from publishers (uplink loss recovery)
+	Publisher bool `yaml:"publisher,omitempty"`
+	// dynamic payload type used for the flexfec-03 codec
+	PayloadType int `yaml:"payload_type,omitempty"`
+	// NumMediaPackets is the number of media packets per FEC protection block for
+	// subscriber-side generation. The repair packets for a block are only emitted
+	// after this many media packets, so smaller blocks recover faster (lower added
+	// latency) but cost proportionally more overhead. Defaults to 10.
+	NumMediaPackets uint32 `yaml:"num_media_packets,omitempty"`
+	// NumFecPackets is the number of repair packets generated per block. Bandwidth
+	// overhead is approximately NumFecPackets/NumMediaPackets, and a block can recover
+	// at most NumFecPackets lost packets out of its NumMediaPackets. Note this is a
+	// fixed (non-adaptive), block-based scheme: it protects scattered loss well but a
+	// burst exceeding NumFecPackets within one block is unrecoverable by FEC.
+	// Defaults to 2 (i.e. ~20% overhead with the default NumMediaPackets).
+	NumFecPackets uint32 `yaml:"num_fec_packets,omitempty"`
+}
+
+func (f FlexFECConfig) Enabled() bool {
+	return (f.Subscriber || f.Publisher) && f.PayloadType != 0
+}
+
+// EncoderParams returns the (numMediaPackets, numFecPackets) used to configure the
+// subscriber-side FlexFEC encoder, applying sane defaults and clamping so the values
+// are always usable (1 <= numFec < numMedia, numMedia >= 2).
+func (f FlexFECConfig) EncoderParams() (numMediaPackets uint32, numFecPackets uint32) {
+	numMediaPackets = f.NumMediaPackets
+	numFecPackets = f.NumFecPackets
+	if numMediaPackets == 0 {
+		numMediaPackets = 10
+	}
+	if numFecPackets == 0 {
+		numFecPackets = 2
+	}
+	if numMediaPackets < 2 {
+		numMediaPackets = 2
+	}
+	if numFecPackets >= numMediaPackets {
+		numFecPackets = numMediaPackets - 1
+	}
+	return numMediaPackets, numFecPackets
 }
 
 type TURNServer struct {
@@ -390,6 +451,13 @@ var DefaultConfig = Config{
 			SendSideBWEPacer:          string(pacer.PacerBehaviorNoQueue),
 			SendSideBWE:               sendsidebwe.DefaultSendSideBWEConfig,
 		},
+		FlexFEC: FlexFECConfig{
+			Subscriber:      false,
+			Publisher:       false,
+			PayloadType:     49,
+			NumMediaPackets: 10,
+			NumFecPackets:   2,
+		},
 	},
 	Audio: sfu.DefaultAudioConfig,
 	Video: VideoConfig{

pkg/rtc/config.go

@@ -37,6 +37,7 @@ type WebRTCConfig struct {
 	Receiver      ReceiverConfig
 	Publisher     DirectionConfig
 	Subscriber    DirectionConfig
+	FlexFEC       config.FlexFECConfig
 }
 
 type ReceiverConfig struct {
@@ -88,6 +89,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) {
 		},
 		Publisher:  getPublisherConfig(false),
 		Subscriber: getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE),
+		FlexFEC:    rtcConf.FlexFEC,
 	}, nil
 }
 

pkg/rtc/mediaengine.go

@@ -86,6 +86,28 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedbac
 	return nil
 }
 
+// flexFEC03CodecParameters returns the flexfec-03 (RFC 8627 draft-03) video codec
+// parameters for the given dynamic payload type. The repair-window fmtp matches the
+// value pion's webrtc.ConfigureFlexFEC03 advertises so publisher and subscriber sides
+// negotiate consistently.
+func flexFEC03CodecParameters(payloadType webrtc.PayloadType) webrtc.RTPCodecParameters {
+	return webrtc.RTPCodecParameters{
+		RTPCodecCapability: webrtc.RTPCodecCapability{
+			MimeType:    webrtc.MimeTypeFlexFEC03,
+			ClockRate:   90000,
+			SDPFmtpLine: "repair-window=10000000",
+		},
+		PayloadType: payloadType,
+	}
+}
+
+// registerFlexFECCodec registers the flexfec-03 codec on the media engine so that the
+// remote peer can negotiate sending a FlexFEC repair stream to the SFU. This is the
+// receive (publisher) side - it does NOT add the encoder interceptor.
+func registerFlexFECCodec(me *webrtc.MediaEngine, payloadType webrtc.PayloadType) error {
+	return me.RegisterCodec(flexFEC03CodecParameters(payloadType), webrtc.RTPCodecTypeVideo)
+}
+
 func registerHeaderExtensions(me *webrtc.MediaEngine, rtpHeaderExtension RTPHeaderExtensionConfig) error {
 	for _, extension := range rtpHeaderExtension.Video {
 		if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeVideo); err != nil {
@@ -178,6 +200,14 @@ func filterCodecs(
 			continue
 		}
 
+		// FlexFEC is an infrastructure codec (like RTX): it only appears in the codec
+		// list when it was registered on the media engine for this peer connection, so
+		// retain it unconditionally. The room enabled-codec list only tracks media codecs.
+		if strings.EqualFold(c.RTPCodecCapability.MimeType, webrtc.MimeTypeFlexFEC03) {
+			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/transport.go

@@ -49,6 +49,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/sfu/bwe/remotebwe"
 	"github.com/livekit/livekit-server/pkg/sfu/bwe/sendsidebwe"
 	"github.com/livekit/livekit-server/pkg/sfu/datachannel"
+	"github.com/livekit/livekit-server/pkg/sfu/flexfec"
 	sfuinterceptor "github.com/livekit/livekit-server/pkg/sfu/interceptor"
 	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 	pd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/playoutdelay"
@@ -443,6 +444,74 @@ func newPeerConnection(
 	se.LoggerFactory = pionlogger.NewLoggerFactory(params.Logger)
 
 	ir := &interceptor.Registry{}
+
+	// FlexFEC (flexfec-03) setup.
+	//
+	// FlexFEC is terminated at the SFU; the two directions are independent:
+	//   - Subscriber side (this PC sends media to the client): register the flexfec-03
+	//     codec AND the pion encoder interceptor so a fresh repair stream is generated
+	//     per downstream. ConfigureFlexFEC03 must be added before any interceptor that
+	//     mutates outgoing RTP (e.g. the TWCC header extension interceptor below).
+	//   - Publisher side (this PC receives media from the client): register only the
+	//     flexfec-03 codec so the client negotiates sending us a repair stream, which the
+	//     buffer layer associates and decodes.
+	if flexFEC := params.Config.FlexFEC; flexFEC.Enabled() {
+		fecPayloadType := webrtc.PayloadType(flexFEC.PayloadType)
+		// in single-PC / one-shot mode a single PC both sends and receives, so Transport is
+		// PUBLISHER while IsSendSide is true - it then needs both behaviors.
+		isSubscriberRole := params.IsSendSide
+		isPublisherRole := params.Transport == livekit.SignalTarget_PUBLISHER
+		wantEncoder := isSubscriberRole && flexFEC.Subscriber
+		wantReceiveCodec := isPublisherRole && flexFEC.Publisher
+		switch {
+		case wantEncoder:
+			// Register the flexfec-03 codec so the sender allocates a repair SSRC and the
+			// stream info carries the FEC payload type / SSRC, then add our own encoder
+			// interceptor. We deliberately do NOT use pion's webrtc.ConfigureFlexFEC03:
+			// pion's encoder interceptor retains each media packet's payload slice by
+			// reference and only XORs the block once it is full, but LiveKit recycles RTP
+			// payload buffers back to a sync.Pool as soon as the packet is written (see
+			// pacer.Base.SendPacket). By the time pion computed the parity, earlier
+			// payloads in the block had been overwritten, producing FEC that protects
+			// garbage and causing receiver-side frame corruption / PLI storms. Our
+			// interceptor copies the payload when buffering to avoid this.
+			//
+			// It is added to the registry here, before the TWCC header extension
+			// interceptor below, so that when the interceptor-based send-side BWE is in
+			// use the generated FEC packets are assigned transport-wide sequence numbers
+			// and accounted for by congestion control.
+			numMediaPackets, numFecPackets := flexFEC.EncoderParams()
+			if err := registerFlexFECCodec(me, fecPayloadType); err != nil {
+				params.Logger.Warnw("failed to register flexfec-03 codec for generation", err)
+			} else {
+				ir.Add(flexfec.NewEncoderInterceptorFactory(numMediaPackets, numFecPackets))
+				params.Logger.Infow("flexfec-03 subscriber generation enabled",
+					"payloadType", fecPayloadType,
+					"numMediaPackets", numMediaPackets,
+					"numFecPackets", numFecPackets,
+				)
+				// FEC packets are generated inside our encoder interceptor, downstream of
+				// LiveKit's pacer. With the RemoteBWE path the added load is observed
+				// through receiver feedback (loss / REMB), and with the interceptor-based
+				// send-side BWE the packets are TWCC-tagged (see ordering above). With
+				// LiveKit's own pacer-based send-side BWE the FEC overhead is not fed back
+				// into the estimator, so warn to make the limitation explicit.
+				if params.CongestionControlConfig.UseSendSideBWE && !params.CongestionControlConfig.UseSendSideBWEInterceptor {
+					params.Logger.Warnw(
+						"flexfec-03 generation enabled with pacer-based send-side BWE; "+
+							"FEC overhead is not accounted for by congestion control", nil,
+					)
+				}
+			}
+		case wantReceiveCodec:
+			if err := registerFlexFECCodec(me, fecPayloadType); err != nil {
+				params.Logger.Warnw("failed to register flexfec-03 codec", err)
+			} else {
+				params.Logger.Debugw("flexfec-03 publisher reception enabled", "payloadType", fecPayloadType)
+			}
+		}
+	}
+
 	if params.IsSendSide {
 		if params.CongestionControlConfig.UseSendSideBWEInterceptor && !params.CongestionControlConfig.UseSendSideBWE {
 			params.Logger.Infow("using send side BWE - interceptor")
@@ -1649,6 +1718,15 @@ func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remo
 				t.params.Config.BufferFactory.SetRTXPair(repair, base, "")
 			}
 		}
+		if t.params.Config.FlexFEC.Publisher {
+			fecRepairs := nonSimulcastFECRepairsFromSDP(parsed, t.params.Logger)
+			if len(fecRepairs) > 0 {
+				t.params.Logger.Debugw("flexfec pairs found from sdp", "ssrcs", fecRepairs)
+				for repair, base := range fecRepairs {
+					t.params.Config.BufferFactory.SetFECPair(repair, base)
+				}
+			}
+		}
 		return nil
 	}
 
@@ -2874,6 +2952,16 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, o
 		}
 	}
 
+	if t.params.Config.FlexFEC.Publisher {
+		fecRepairs := nonSimulcastFECRepairsFromSDP(parsed, t.params.Logger)
+		if len(fecRepairs) > 0 {
+			t.params.Logger.Debugw("flexfec pairs found from sdp", "ssrcs", fecRepairs)
+			for repair, base := range fecRepairs {
+				t.params.Config.BufferFactory.SetFECPair(repair, base)
+			}
+		}
+	}
+
 	if t.currentOfferIceCredential == "" || offerRestartICE {
 		t.currentOfferIceCredential = iceCredential
 	}
@@ -3245,6 +3333,47 @@ func nonSimulcastRTXRepairsFromSDP(s *sdp.SessionDescription, logger logger.Logg
 	return rtxRepairFlows
 }
 
+// nonSimulcastFECRepairsFromSDP extracts FlexFEC repair flows from non-simulcast media
+// sections, returning a map of FEC(repair) SSRC -> source(base) SSRC. It mirrors
+// nonSimulcastRTXRepairsFromSDP but parses `a=ssrc-group:FEC-FR <source> <fec>` lines
+// (RFC 5956), where the first SSRC is the protected source stream and the second is the
+// FlexFEC repair stream.
+func nonSimulcastFECRepairsFromSDP(s *sdp.SessionDescription, logger logger.Logger) map[uint32]uint32 {
+	fecRepairFlows := map[uint32]uint32{}
+	for _, media := range s.MediaDescriptions {
+		var ridFound bool
+		fecPairs := make(map[uint32]uint32)
+	findFEC:
+		for _, attr := range media.Attributes {
+			switch attr.Key {
+			case "rid":
+				ridFound = true
+				break findFEC
+			case sdp.AttrKeySSRCGroup:
+				split := strings.Split(attr.Value, " ")
+				if split[0] == sdp.SemanticTokenForwardErrorCorrectionFramework && len(split) == 3 {
+					baseSsrc, err := strconv.ParseUint(split[1], 10, 32)
+					if err != nil {
+						logger.Warnw("Failed to parse FEC source SSRC", err, "ssrc", split[1])
+						continue
+					}
+					fecRepairFlow, err := strconv.ParseUint(split[2], 10, 32)
+					if err != nil {
+						logger.Warnw("Failed to parse FEC repair SSRC", err, "ssrc", split[2])
+						continue
+					}
+					fecPairs[uint32(fecRepairFlow)] = uint32(baseSsrc)
+				}
+			}
+		}
+		if !ridFound {
+			maps.Copy(fecRepairFlows, fecPairs)
+		}
+	}
+
+	return fecRepairFlows
+}
+
 // ----------------------
 
 type iceCandidatePairStatsEncoder struct {

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"
@@ -33,6 +34,9 @@ import (
 const (
 	rtcpReceiverReportDelta = 1e9
 
+	// fecReportInterval bounds how often a buffer logs a FlexFEC recovery summary.
+	fecReportInterval = 5e9
+
 	InitPacketBufferSizeVideo = 300
 	InitPacketBufferSizeAudio = 70
 )
@@ -72,6 +76,28 @@ type Buffer struct {
 
 	primaryBufferForRTX *Buffer
 	rtxPktBuf           []byte
+
+	// FlexFEC (publisher -> SFU recovery).
+	// primaryBufferForFEC is set on a repair (FEC) buffer and points at the source
+	// buffer it protects. fecDecoder lives on the source buffer and reconstructs lost
+	// source packets from the repair stream. fecPktBuf is scratch for re-marshaling
+	// recovered packets before injecting them back into the source buffer.
+	primaryBufferForFEC *Buffer
+	fecDecoder          *flexfec.Decoder
+	fecPktBuf           []byte
+
+	// FlexFEC observability. Counters are cumulative for the lifetime of the source
+	// buffer; the *Reported fields snapshot the last logged values so the periodic
+	// summary can report deltas.
+	fecPacketsReceived   uint64
+	fecPacketsRecovered  uint64
+	fecReportedAt        int64
+	fecReportedReceived  uint64
+	fecReportedRecovered uint64
+
+	// Debug-only simulated publisher ("robot") uplink impairment; nil unless the
+	// LK_UPLINK_* env vars are set. See buffer_impair.go.
+	impair *uplinkImpair
 }
 
 func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer {
@@ -84,6 +110,7 @@ func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer {
 		SendPLI:            b.sendPLI,
 		IsReportingEnabled: true,
 	})
+	b.initUplinkImpair()
 	return b
 }
 
@@ -138,7 +165,24 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 }
 
 // Write adds an RTP Packet, ordering is not guaranteed, newer packets may arrive later
-func (b *Buffer) Write(pkt []byte) (n int, err error) {
+// Write is the inbound entry point for the publisher's RTP (media, RTX, and FEC SSRCs).
+// It applies the optional debug uplink impairment (loss + one-way delay) before handing
+// the packet to writeNow; both are no-ops unless the LK_UPLINK_* env vars are set.
+func (b *Buffer) Write(pkt []byte) (int, error) {
+	if b.impair != nil {
+		if b.impair.dropInbound() {
+			// Packet "lost" on the simulated 5G uplink; report success so pion's read
+			// loop is unaffected.
+			return len(pkt), nil
+		}
+		if b.impair.delayInbound(b, pkt) {
+			return len(pkt), nil
+		}
+	}
+	return b.writeNow(pkt)
+}
+
+func (b *Buffer) writeNow(pkt []byte) (n int, err error) {
 	var rtpPacket rtp.Packet
 	err = rtpPacket.Unmarshal(pkt)
 	if err != nil {
@@ -179,6 +223,13 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) {
 		return
 	}
 
+	// handle FlexFEC repair packet
+	if pb := b.primaryBufferForFEC; pb != nil {
+		b.Unlock()
+		pb.writeFEC(&rtpPacket, now)
+		return
+	}
+
 	if !b.isBound {
 		packet := make([]byte, len(pkt))
 		copy(packet, pkt)
@@ -203,16 +254,33 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) {
 	}
 
 	rtcpPackets := b.calc(pkt, &rtpPacket, now, false, false)
+
+	// feed received source packets to the FlexFEC decoder so it can reconstruct
+	// previously lost packets, and inject any recoveries back into this buffer.
+	if b.fecDecoder != nil {
+		b.injectRecoveredLocked(b.fecDecoder.Decode(rtpPacket), now)
+	}
+
 	b.Unlock()
 
 	if len(rtcpPackets) != 0 {
-		if cb := b.getOnRtcpFeedback(); cb != nil {
-			cb(rtcpPackets)
-		}
+		b.emitRTCPFeedback(rtcpPackets)
 	}
 	return
 }
 
+// emitRTCPFeedback delivers SFU->publisher feedback (NACK/RR/etc). When the debug uplink
+// delay is set it defers delivery by the one-way delay so a NACK->RTX recovery costs a
+// full round trip; otherwise it fires immediately.
+func (b *Buffer) emitRTCPFeedback(rtcpPackets []rtcp.Packet) {
+	if b.impair != nil && b.impair.delayFeedback(b, rtcpPackets) {
+		return
+	}
+	if cb := b.getOnRtcpFeedback(); cb != nil {
+		cb(rtcpPackets)
+	}
+}
+
 func (b *Buffer) SetPrimaryBufferForRTX(primaryBuffer *Buffer) {
 	b.Lock()
 	b.primaryBufferForRTX = primaryBuffer
@@ -233,6 +301,114 @@ func (b *Buffer) SetPrimaryBufferForRTX(primaryBuffer *Buffer) {
 	}
 }
 
+// SetPrimaryBufferForFEC associates this (repair/FEC) buffer with the source buffer it
+// protects. It installs a FlexFEC decoder on the source buffer and replays any FEC
+// packets that arrived before the association was known.
+func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) {
+	primaryBuffer.enableFECDecoder(b.BufferBase.SSRC())
+
+	b.Lock()
+	b.primaryBufferForFEC = primaryBuffer
+	pkts := b.pPackets
+	b.pPackets = nil
+	b.Unlock()
+
+	for _, pp := range pkts {
+		var rtpPacket rtp.Packet
+		if err := rtpPacket.Unmarshal(pp.packet); err != nil {
+			continue
+		}
+		primaryBuffer.writeFEC(&rtpPacket, pp.arrivalTime)
+	}
+}
+
+// enableFECDecoder installs a FlexFEC decoder on this (source) buffer for the given
+// repair SSRC. Safe to call before the buffer is bound.
+func (b *Buffer) enableFECDecoder(fecSSRC uint32) {
+	b.Lock()
+	defer b.Unlock()
+
+	if b.fecDecoder == nil {
+		b.fecDecoder = flexfec.NewDecoder(fecSSRC, b.BufferBase.SSRC())
+		b.logger.Infow("flexfec decoder enabled", "fecSSRC", fecSSRC, "mediaSSRC", b.BufferBase.SSRC())
+	}
+}
+
+// writeFEC feeds a received FlexFEC repair packet to this (source) buffer's decoder and
+// injects any recovered source packets back into the buffer. Called on the source buffer.
+func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) {
+	b.Lock()
+	defer b.Unlock()
+
+	if b.fecDecoder == nil {
+		return
+	}
+
+	b.fecPacketsReceived++
+	b.injectRecoveredLocked(b.fecDecoder.Decode(*fecPkt), arrivalTime)
+	b.maybeLogFECStatsLocked(arrivalTime)
+}
+
+// maybeLogFECStatsLocked emits a periodic FlexFEC recovery summary at most once per
+// fecReportInterval, and only when there has been FEC activity since the last summary.
+// b.Lock must be held.
+func (b *Buffer) maybeLogFECStatsLocked(now int64) {
+	if b.fecReportedAt == 0 {
+		b.fecReportedAt = now
+		b.fecReportedReceived = b.fecPacketsReceived
+		b.fecReportedRecovered = b.fecPacketsRecovered
+		return
+	}
+
+	if now-b.fecReportedAt < fecReportInterval {
+		return
+	}
+
+	receivedDelta := b.fecPacketsReceived - b.fecReportedReceived
+	recoveredDelta := b.fecPacketsRecovered - b.fecReportedRecovered
+
+	b.fecReportedAt = now
+	b.fecReportedReceived = b.fecPacketsReceived
+	b.fecReportedRecovered = b.fecPacketsRecovered
+
+	if receivedDelta == 0 && recoveredDelta == 0 {
+		return
+	}
+
+	b.logger.Infow(
+		"flexfec recovery stats",
+		"mediaSSRC", b.BufferBase.SSRC(),
+		"fecPacketsReceived", b.fecPacketsReceived,
+		"fecPacketsReceivedDelta", receivedDelta,
+		"packetsRecovered", b.fecPacketsRecovered,
+		"packetsRecoveredDelta", recoveredDelta,
+	)
+}
+
+// injectRecoveredLocked re-injects FlexFEC-recovered source packets into the buffer as
+// repaired packets (treated like RTX repairs, so they become NACK/forward eligible).
+// b.Lock must be held.
+func (b *Buffer) injectRecoveredLocked(recovered []rtp.Packet, arrivalTime int64) {
+	if len(recovered) == 0 || !b.isBound {
+		return
+	}
+
+	if b.fecPktBuf == nil {
+		b.fecPktBuf = make([]byte, bucket.RTPMaxPktSize)
+	}
+
+	for i := range recovered {
+		pkt := recovered[i]
+		n, err := pkt.MarshalTo(b.fecPktBuf)
+		if err != nil {
+			b.logger.Errorw("could not marshal flexfec recovered packet", err, "sn", pkt.SequenceNumber)
+			continue
+		}
+		b.calc(b.fecPktBuf[:n], &pkt, arrivalTime, false, true)
+		b.fecPacketsRecovered++
+	}
+}
+
 func (b *Buffer) NotifyRTX(ssrc uint32, repairSSRC uint32, rsid string) {
 	if onNotifyRTX := b.getOnNotifyRTX(); onNotifyRTX != nil {
 		onNotifyRTX(ssrc, repairSSRC, rsid)

pkg/sfu/buffer/buffer_impair.go

@@ -0,0 +1,204 @@
+// Copyright 2024 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"
+	"os"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/pion/rtcp"
+)
+
+// Debug-only simulated publisher ("robot") uplink impairment for the local FlexFEC test
+// harness (scripts/flexfec). It models the lossy, higher-RTT first hop a teleop robot sees
+// on a wireless/5G link: random loss on inbound RTP (media + RTX + FEC, since real loss
+// hits everything) and a one-way delay applied symmetrically to inbound media AND the
+// SFU->publisher feedback (NACK), so a NACK->RTX recovery costs a full round trip while a
+// normal packet pays one one-way delay. Publisher-side FlexFEC repairs that loss inline at
+// the SFU without any round trip -- which is the latency it saves here.
+//
+// CRITICAL ORDERING NOTE: a real link is ONE pipe -- it delays/loses every packet of every
+// SSRC (media, RTX, FEC) together while preserving their global arrival order. The SFU's
+// FlexFEC decoder depends on that ordering (it XORs a repair packet against the surviving
+// media packets of its block). An earlier design gave each Buffer (media/RTX/FEC SSRC) its
+// own delay goroutine; the independent goroutines reordered the streams relative to each
+// other and FEC recovery dropped to zero. So this impairment is a single process-global
+// FIFO delay line with one drain goroutine: with a constant delay, FIFO dequeue preserves
+// the exact arrival order (just time-shifted), and recovery behaves identically to the
+// no-delay path.
+//
+// Controlled by env vars; absent/zero values make this a complete no-op:
+//
+//	LK_PUB_LOSS      fractional loss on the publisher->SFU path, e.g. 0.03 (3%)
+//	LK_PUB_DELAY_MS  one-way publisher<->SFU delay in ms (inbound media + outbound NACK)
+//
+// (These are the robot-link knobs. The operator/subscriber-side knobs live in the sfu
+// package: LK_DOWNLINK_LOSS, LK_DOWNLINK_DELAY_MS, LK_UPLINK_DELAY_MS.)
+const (
+	envPubLoss    = "LK_PUB_LOSS"
+	envPubDelayMs = "LK_PUB_DELAY_MS"
+
+	impairQueueDepth = 8192
+)
+
+type delayedInbound struct {
+	b         *Buffer
+	pkt       []byte
+	releaseAt int64 // unix nanos
+}
+
+type delayedFeedback struct {
+	b         *Buffer
+	pkts      []rtcp.Packet
+	releaseAt int64 // unix nanos
+}
+
+type uplinkImpair struct {
+	loss  float64
+	delay time.Duration
+
+	rngMu sync.Mutex
+	rng   *rand.Rand
+
+	inQueue chan delayedInbound
+	fbQueue chan delayedFeedback
+}
+
+var (
+	globalImpairOnce sync.Once
+	globalImpair     *uplinkImpair
+)
+
+// getUplinkImpair lazily builds the process-global uplink impairment from the LK_PUB_* env
+// vars, or returns nil when none are set. All inbound Buffers share the one instance so the
+// delay line preserves global packet ordering across every SSRC (see the note above).
+func getUplinkImpair() *uplinkImpair {
+	globalImpairOnce.Do(func() {
+		loss := parseEnvFloat(envPubLoss)
+		delay := time.Duration(parseEnvInt(envPubDelayMs)) * time.Millisecond
+		if loss <= 0 && delay <= 0 {
+			return
+		}
+		imp := &uplinkImpair{
+			loss:  loss,
+			delay: delay,
+			rng:   rand.New(rand.NewSource(time.Now().UnixNano())),
+		}
+		if delay > 0 {
+			imp.inQueue = make(chan delayedInbound, impairQueueDepth)
+			imp.fbQueue = make(chan delayedFeedback, impairQueueDepth)
+			go imp.runInboundDelayLine()
+			go imp.runFeedbackDelayLine()
+		}
+		globalImpair = imp
+	})
+	return globalImpair
+}
+
+// initUplinkImpair attaches the shared uplink impairment to this Buffer (nil unless the
+// LK_PUB_* env vars are set). Called once per Buffer from NewBuffer.
+func (b *Buffer) initUplinkImpair() {
+	b.impair = getUplinkImpair()
+	if b.impair != nil && b.logger != nil {
+		b.logger.Infow("uplink impairment enabled (debug)", "lossFraction", b.impair.loss, "delay", b.impair.delay)
+	}
+}
+
+func (u *uplinkImpair) dropInbound() bool {
+	if u.loss <= 0 {
+		return false
+	}
+	u.rngMu.Lock()
+	r := u.rng.Float64()
+	u.rngMu.Unlock()
+	return r < u.loss
+}
+
+// delayInbound returns true when the packet was deferred onto the shared delay line (caller
+// should return immediately); false when no delay is configured and the caller should
+// proceed inline.
+func (u *uplinkImpair) delayInbound(b *Buffer, pkt []byte) bool {
+	if u.delay <= 0 {
+		return false
+	}
+	// pion recycles pkt after Write returns, so copy before deferring.
+	cp := make([]byte, len(pkt))
+	copy(cp, pkt)
+	item := delayedInbound{b: b, pkt: cp, releaseAt: time.Now().Add(u.delay).UnixNano()}
+	select {
+	case u.inQueue <- item:
+	default:
+		// Queue full: drop rather than block pion's read loop.
+	}
+	return true
+}
+
+// delayFeedback returns true when feedback was deferred onto the shared delay line; false
+// when no delay is configured and the caller should send immediately.
+func (u *uplinkImpair) delayFeedback(b *Buffer, pkts []rtcp.Packet) bool {
+	if u.delay <= 0 {
+		return false
+	}
+	item := delayedFeedback{b: b, pkts: pkts, releaseAt: time.Now().Add(u.delay).UnixNano()}
+	select {
+	case u.fbQueue <- item:
+	default:
+	}
+	return true
+}
+
+// runInboundDelayLine releases delayed inbound packets, in global FIFO order, into the
+// owning buffer's writeNow. A constant delay keeps releaseAt monotonic, so a single
+// goroutine preserves the exact arrival order across all SSRCs.
+func (u *uplinkImpair) runInboundDelayLine() {
+	for item := range u.inQueue {
+		if w := item.releaseAt - time.Now().UnixNano(); w > 0 {
+			time.Sleep(time.Duration(w))
+		}
+		_, _ = item.b.writeNow(item.pkt)
+	}
+}
+
+// runFeedbackDelayLine releases delayed SFU->publisher feedback (NACK/RR) after the one-way
+// uplink delay.
+func (u *uplinkImpair) runFeedbackDelayLine() {
+	for item := range u.fbQueue {
+		if w := item.releaseAt - time.Now().UnixNano(); w > 0 {
+			time.Sleep(time.Duration(w))
+		}
+		if cb := item.b.getOnRtcpFeedback(); cb != nil {
+			cb(item.pkts)
+		}
+	}
+}
+
+func parseEnvFloat(key string) float64 {
+	v, err := strconv.ParseFloat(os.Getenv(key), 64)
+	if err != nil {
+		return 0
+	}
+	return v
+}
+
+func parseEnvInt(key string) int {
+	v, err := strconv.Atoi(os.Getenv(key))
+	if err != nil {
+		return 0
+	}
+	return v
+}

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 // flexfec repair -> base
 }
 
 func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser {
@@ -89,10 +91,24 @@ func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io
 				break
 			}
 		}
+		for repair, base := range f.fecPair {
+			if repair == ssrc {
+				if baseBuffer, ok := f.rtpBuffers[base]; ok {
+					buffer.SetPrimaryBufferForFEC(baseBuffer)
+				}
+				break
+			} else if base == ssrc {
+				if repairBuffer, ok := f.rtpBuffers[repair]; ok {
+					repairBuffer.SetPrimaryBufferForFEC(buffer)
+				}
+				break
+			}
+		}
 		buffer.OnClose(func() {
 			f.Lock()
 			delete(f.rtpBuffers, ssrc)
 			delete(f.rtxPair, ssrc)
+			delete(f.fecPair, ssrc)
 			f.Unlock()
 		})
 		return buffer
@@ -132,3 +148,19 @@ func (f *Factory) SetRTXPair(repair, base uint32, rsid string) {
 		}
 	}
 }
+
+// SetFECPair associates a FlexFEC repair stream SSRC with the source (base) SSRC it
+// protects, as signalled by an a=ssrc-group:FEC-FR line. If both buffers already exist
+// the association is wired immediately, otherwise it is remembered and applied when the
+// missing buffer is created.
+func (f *Factory) SetFECPair(repair, base uint32) {
+	f.Lock()
+	repairBuffer, baseBuffer := f.rtpBuffers[repair], f.rtpBuffers[base]
+	if repairBuffer == nil || baseBuffer == nil {
+		f.fecPair[repair] = base
+	}
+	f.Unlock()
+	if repairBuffer != nil && baseBuffer != nil {
+		repairBuffer.SetPrimaryBufferForFEC(baseBuffer)
+	}
+}

pkg/sfu/buffer/flexfec_test.go

@@ -0,0 +1,148 @@
+// Copyright 2024 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 (
+	"testing"
+	"time"
+
+	"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/require"
+)
+
+// TestFlexFECRecoveryThroughFactory exercises the full SFU receive path: the source and
+// repair buffers are created through the factory (as the SRTP layer would), associated
+// via SetFECPair, and a lost source packet is recovered from the FlexFEC repair stream
+// and forwarded out of the source buffer.
+func TestFlexFECRecoveryThroughFactory(t *testing.T) {
+	const (
+		mediaSSRC = uint32(0xAABBCCDD)
+		fecSSRC   = uint32(0x11223344)
+		baseSeq   = uint16(5000)
+		numMedia  = 10
+		dropIdx   = 6
+	)
+
+	factory := NewFactoryOfBufferFactory(InitPacketBufferSizeVideo, InitPacketBufferSizeAudio).CreateBufferFactory()
+
+	mediaBuf, ok := factory.GetOrNew(packetio.RTPBufferPacket, mediaSSRC).(*Buffer)
+	require.True(t, ok)
+	mediaBuf.codecType = webrtc.RTPCodecTypeAudio
+	require.NoError(t, mediaBuf.Bind(
+		webrtc.RTPParameters{Codecs: []webrtc.RTPCodecParameters{opusCodec}},
+		opusCodec.RTPCodecCapability,
+		0,
+	))
+
+	fecBuf, ok := factory.GetOrNew(packetio.RTPBufferPacket, fecSSRC).(*Buffer)
+	require.True(t, ok)
+
+	// associate the repair stream with the source stream (as a=ssrc-group:FEC-FR would)
+	factory.SetFECPair(fecSSRC, mediaSSRC)
+	require.NotNil(t, mediaBuf.fecDecoder)
+	require.Equal(t, mediaBuf, fecBuf.primaryBufferForFEC)
+
+	// build media packets and the protecting FlexFEC packet
+	media := make([]rtp.Packet, numMedia)
+	for i := 0; i < numMedia; i++ {
+		payload := make([]byte, 16)
+		for j := range payload {
+			payload[j] = byte((i*11 + j*5 + 1) & 0xff)
+		}
+		media[i] = rtp.Packet{
+			Header: rtp.Header{
+				Version:        2,
+				PayloadType:    uint8(opusCodec.PayloadType),
+				SequenceNumber: baseSeq + uint16(i),
+				Timestamp:      uint32(8000 + i*960),
+				SSRC:           mediaSSRC,
+			},
+			Payload: payload,
+		}
+	}
+
+	encoder := flexfec.NewFlexEncoder03(uint8(49), fecSSRC)
+	mediaForFec := make([]rtp.Packet, numMedia)
+	copy(mediaForFec, media)
+	fecPackets := encoder.EncodeFec(mediaForFec, 1)
+	require.NotEmpty(t, fecPackets)
+
+	// collect forwarded packets out of the source buffer
+	got := make(chan *ExtPacket, numMedia*2)
+	go func() {
+		var buf [1500]byte
+		for {
+			ep, err := mediaBuf.ReadExtended(buf[:])
+			if err != nil {
+				return
+			}
+			if ep != nil {
+				clone := *ep
+				got <- &clone
+			}
+		}
+	}()
+
+	// deliver all source packets except the dropped one
+	for i := 0; i < numMedia; i++ {
+		if i == dropIdx {
+			continue
+		}
+		raw, err := media[i].Marshal()
+		require.NoError(t, err)
+		_, err = mediaBuf.Write(raw)
+		require.NoError(t, err)
+	}
+
+	// deliver the repair packet to the FEC buffer -> triggers recovery + injection
+	for _, fp := range fecPackets {
+		raw, err := fp.Marshal()
+		require.NoError(t, err)
+		_, err = fecBuf.Write(raw)
+		require.NoError(t, err)
+	}
+
+	// the dropped source packet must be recovered and forwarded
+	recoveredSeq := baseSeq + uint16(dropIdx)
+	deadline := time.After(2 * time.Second)
+	seen := map[uint16]*ExtPacket{}
+	for {
+		select {
+		case ep := <-got:
+			seen[ep.Packet.SequenceNumber] = ep
+			if rec, found := seen[recoveredSeq]; found {
+				require.Equal(t, mediaSSRC, rec.Packet.SSRC)
+				require.Equal(t, media[dropIdx].Payload, rec.Packet.Payload)
+				require.Equal(t, media[dropIdx].Timestamp, rec.Packet.Timestamp)
+				_ = mediaBuf.Close()
+				return
+			}
+		case <-deadline:
+			_ = mediaBuf.Close()
+			t.Fatalf("recovered packet seq %d not forwarded; saw %v", recoveredSeq, seqKeys(seen))
+		}
+	}
+}
+
+func seqKeys(m map[uint16]*ExtPacket) []uint16 {
+	keys := make([]uint16, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	return keys
+}

pkg/sfu/downtrack.go

@@ -630,18 +630,22 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
 		)
 		d.params.Logger.Debugw("DownTrack.Bind", logFields...)
 
-		d.writeStream = t.WriteStream()
+		// maybeWrapDownlinkImpairment is a no-op unless the LK_DOWNLINK_* debug env vars
+		// are set (local FlexFEC test harness only); see downtrack_impair.go.
+		d.writeStream = maybeWrapDownlinkImpairment(t.WriteStream(), d.params.Logger)
 		if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrc).(*buffer.RTCPReader); rr != nil {
-			rr.OnPacket(func(pkt []byte) {
+			// maybeDelayInboundRTCP is a no-op unless the LK_UPLINK_DELAY_MS debug env var
+			// is set (local FlexFEC test harness only); see downtrack_impair.go.
+			rr.OnPacket(maybeDelayInboundRTCP(func(pkt []byte) {
 				d.handleRTCP(pkt)
-			})
+			}, d.params.Logger))
 			d.rtcpReader = rr
 		}
 		if d.ssrcRTX != 0 {
 			if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrcRTX).(*buffer.RTCPReader); rr != nil {
-				rr.OnPacket(func(pkt []byte) {
+				rr.OnPacket(maybeDelayInboundRTCP(func(pkt []byte) {
 					d.handleRTCPRTX(pkt)
-				})
+				}, d.params.Logger))
 				d.rtcpReaderRTX = rr
 			}
 		}

pkg/sfu/downtrack_impair.go

@@ -0,0 +1,231 @@
+// Copyright 2024 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sfu
+
+import (
+	"math/rand"
+	"os"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/pion/rtp"
+	"github.com/pion/webrtc/v4"
+
+	"github.com/livekit/protocol/logger"
+)
+
+// Debug-only downlink network impairment for the local FlexFEC test harness
+// (scripts/flexfec). When the SFU, publisher and subscriber all run on one macOS box,
+// media never leaves the host, and pf/dummynet (netem.sh) cannot shape same-host UDP
+// because the kernel short-circuits local delivery before the pf OUTPUT hook. To run a
+// meaningful FEC-vs-RTX comparison we instead inject loss + one-way delay in software,
+// right at the subscriber DownTrack egress.
+//
+// Controlled entirely by env vars; absent/zero values make wrapping a no-op so this has
+// ZERO effect on normal operation:
+//
+//	LK_DOWNLINK_LOSS      fractional packet loss applied to SFU->subscriber, e.g. 0.01 (1%)
+//	LK_DOWNLINK_DELAY_MS  one-way delay added to SFU->subscriber in milliseconds, e.g. 40
+//	LK_UPLINK_DELAY_MS    one-way delay added to the subscriber->SFU feedback path (inbound
+//	                      RTCP, i.e. NACK/RR/REMB/PLI) in milliseconds, e.g. 40
+//
+// Set DOWNLINK and UPLINK to the same value to model a symmetric link: a normal frame pays
+// one one-way downlink delay, whereas a NACK->RTX recovery pays uplink (delayed NACK) +
+// downlink (delayed RTX) = a full RTT. That extra round trip is precisely the latency
+// FlexFEC avoids by repairing inline, so a symmetric delay is the fair test of FEC's value.
+const (
+	envDownlinkLoss    = "LK_DOWNLINK_LOSS"
+	envDownlinkDelayMs = "LK_DOWNLINK_DELAY_MS"
+	envUplinkDelayMs   = "LK_UPLINK_DELAY_MS"
+
+	// Bounded queue so a stalled subscriber can't grow memory without limit; at ~300
+	// pkt/s and tens of ms of delay only a handful are ever in flight.
+	impairedQueueDepth = 2048
+)
+
+// maybeWrapDownlinkImpairment returns w wrapped with loss/delay impairment when the
+// LK_DOWNLINK_* env vars request it, otherwise returns w unchanged.
+func maybeWrapDownlinkImpairment(w webrtc.TrackLocalWriter, log logger.Logger) webrtc.TrackLocalWriter {
+	loss := parseEnvFloat(envDownlinkLoss)
+	delay := time.Duration(parseEnvInt(envDownlinkDelayMs)) * time.Millisecond
+	if loss <= 0 && delay <= 0 {
+		return w
+	}
+	if log != nil {
+		log.Infow("downlink impairment enabled (debug)", "lossFraction", loss, "delay", delay)
+	}
+	iw := &impairedWriter{
+		inner: w,
+		loss:  loss,
+		delay: delay,
+		rng:   rand.New(rand.NewSource(time.Now().UnixNano())),
+	}
+	if delay > 0 {
+		iw.queue = make(chan delayedWrite, impairedQueueDepth)
+		go iw.runDelayLine()
+	}
+	return iw
+}
+
+func parseEnvFloat(key string) float64 {
+	v, err := strconv.ParseFloat(os.Getenv(key), 64)
+	if err != nil {
+		return 0
+	}
+	return v
+}
+
+func parseEnvInt(key string) int {
+	v, err := strconv.Atoi(os.Getenv(key))
+	if err != nil {
+		return 0
+	}
+	return v
+}
+
+type delayedWrite struct {
+	header    *rtp.Header
+	payload   []byte
+	raw       []byte
+	releaseAt time.Time
+}
+
+type impairedWriter struct {
+	inner webrtc.TrackLocalWriter
+	loss  float64
+	delay time.Duration
+
+	mu  sync.Mutex
+	rng *rand.Rand
+
+	queue chan delayedWrite
+}
+
+func (iw *impairedWriter) drop() bool {
+	if iw.loss <= 0 {
+		return false
+	}
+	iw.mu.Lock()
+	r := iw.rng.Float64()
+	iw.mu.Unlock()
+	return r < iw.loss
+}
+
+func (iw *impairedWriter) WriteRTP(header *rtp.Header, payload []byte) (int, error) {
+	if iw.drop() {
+		// Report success so DownTrack stats/sequencing are unaffected; the packet is
+		// simply "lost" in the simulated network.
+		return header.MarshalSize() + len(payload), nil
+	}
+	if iw.delay <= 0 {
+		return iw.inner.WriteRTP(header, payload)
+	}
+
+	// The caller's header/payload come from pooled buffers that are recycled the moment
+	// this returns, so everything retained past this call must be copied.
+	dw := delayedWrite{
+		header:    cloneRTPHeader(header),
+		payload:   append([]byte(nil), payload...),
+		releaseAt: time.Now().Add(iw.delay),
+	}
+	n := header.MarshalSize() + len(payload)
+	select {
+	case iw.queue <- dw:
+	default:
+		// Queue full (subscriber stalled): drop rather than block the write path.
+	}
+	return n, nil
+}
+
+func (iw *impairedWriter) Write(b []byte) (int, error) {
+	if iw.drop() {
+		return len(b), nil
+	}
+	if iw.delay <= 0 {
+		return iw.inner.Write(b)
+	}
+	dw := delayedWrite{
+		raw:       append([]byte(nil), b...),
+		releaseAt: time.Now().Add(iw.delay),
+	}
+	select {
+	case iw.queue <- dw:
+	default:
+	}
+	return len(b), nil
+}
+
+// runDelayLine releases queued writes in order once their release time arrives. A constant
+// delay keeps releaseAt monotonic, so a single goroutine preserves packet order.
+func (iw *impairedWriter) runDelayLine() {
+	for dw := range iw.queue {
+		if d := time.Until(dw.releaseAt); d > 0 {
+			time.Sleep(d)
+		}
+		if dw.raw != nil {
+			_, _ = iw.inner.Write(dw.raw)
+			continue
+		}
+		_, _ = iw.inner.WriteRTP(dw.header, dw.payload)
+	}
+}
+
+func cloneRTPHeader(h *rtp.Header) *rtp.Header {
+	// rtp.Header.Clone deep-copies CSRC and extension payloads, so the result is safe to
+	// retain after the caller recycles its pooled buffers.
+	c := h.Clone()
+	return &c
+}
+
+// maybeDelayInboundRTCP wraps an RTCP packet handler so that inbound RTCP (the
+// subscriber->SFU feedback path: NACK/RR/REMB/PLI) is processed LK_UPLINK_DELAY_MS later,
+// modelling the uplink leg of the round trip. With this and a matching downlink delay, a
+// NACK->RTX recovery pays a full RTT while normal media pays only one one-way delay.
+// Returns handler unchanged when LK_UPLINK_DELAY_MS is unset/zero.
+func maybeDelayInboundRTCP(handler func([]byte), log logger.Logger) func([]byte) {
+	delay := time.Duration(parseEnvInt(envUplinkDelayMs)) * time.Millisecond
+	if delay <= 0 {
+		return handler
+	}
+	if log != nil {
+		log.Infow("uplink RTCP delay enabled (debug)", "delay", delay)
+	}
+	queue := make(chan delayedRTCP, impairedQueueDepth)
+	go func() {
+		for d := range queue {
+			if w := time.Until(d.releaseAt); w > 0 {
+				time.Sleep(w)
+			}
+			handler(d.pkt)
+		}
+	}()
+	return func(pkt []byte) {
+		// The reader recycles pkt after this returns, so copy before deferring.
+		dr := delayedRTCP{
+			pkt:       append([]byte(nil), pkt...),
+			releaseAt: time.Now().Add(delay),
+		}
+		select {
+		case queue <- dr:
+		default:
+		}
+	}
+}
+
+type delayedRTCP struct {
+	pkt       []byte
+	releaseAt time.Time
+}

pkg/sfu/flexfec/decoder.go

@@ -0,0 +1,482 @@
+// Copyright 2024 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 by the SFU to recover
+// RTP packets lost on the publisher -> SFU path.
+//
+// FlexFEC (RFC 8627 / draft-ietf-payload-flexible-fec-scheme-03) sends repair
+// packets on a dedicated SSRC. Each repair packet references the protected
+// source packets via a base sequence number plus a bitmask and carries the XOR
+// of the protected packets' headers and payloads. When a single protected
+// packet is missing, it can be reconstructed by XOR-ing the repair packet with
+// the other protected source packets.
+//
+// This is a port of pion/interceptor's (unexported) FlexFEC-03 decoder, adapted
+// to be reusable from the SFU buffer layer. The decoder is NOT safe for
+// concurrent use; the caller must serialize access (the SFU feeds it from a
+// single SRTP read goroutine).
+package flexfec
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"sort"
+
+	"github.com/pion/rtp"
+)
+
+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")
+)
+
+const (
+	defaultMaxMediaPackets = 100
+	defaultMaxFECPackets   = 100
+	recoveredPacketLimit   = 192
+)
+
+// Decoder reconstructs lost source RTP packets from a FlexFEC-03 repair stream.
+//
+// Feed it both the received source packets (SSRC == ProtectedSSRC) and the
+// received repair packets (SSRC == SSRC) via Decode. Each call returns any
+// source packets recovered as a result of the newly inserted packet.
+type Decoder struct {
+	ssrc                uint32
+	protectedStreamSSRC uint32
+	maxMediaPackets     int
+	maxFECPackets       int
+	recoveredPackets    []rtp.Packet
+	receivedFECPackets  []fecPacketState
+}
+
+// NewDecoder creates a decoder for a single (repair SSRC, protected SSRC) pair.
+func NewDecoder(fecSSRC uint32, protectedStreamSSRC uint32) *Decoder {
+	return &Decoder{
+		ssrc:                fecSSRC,
+		protectedStreamSSRC: protectedStreamSSRC,
+		maxMediaPackets:     defaultMaxMediaPackets,
+		maxFECPackets:       defaultMaxFECPackets,
+		recoveredPackets:    make([]rtp.Packet, 0),
+		receivedFECPackets:  make([]fecPacketState, 0),
+	}
+}
+
+// SSRC returns the FlexFEC repair stream SSRC this decoder handles.
+func (d *Decoder) SSRC() uint32 {
+	return d.ssrc
+}
+
+// ProtectedSSRC returns the source stream SSRC this decoder protects.
+func (d *Decoder) ProtectedSSRC() uint32 {
+	return d.protectedStreamSSRC
+}
+
+// Decode inserts a received packet (either a source packet on ProtectedSSRC or
+// a repair packet on SSRC) and returns any source packets recovered as a
+// result. The supplied packet is cloned, so the caller may reuse the backing
+// buffer after Decode returns.
+func (d *Decoder) Decode(receivedPacket rtp.Packet) []rtp.Packet {
+	if receivedPacket.SSRC != d.ssrc && receivedPacket.SSRC != d.protectedStreamSSRC {
+		return nil
+	}
+
+	pkt := clonePacket(receivedPacket)
+
+	if len(d.recoveredPackets) == d.maxMediaPackets {
+		backRecoveredPacket := d.recoveredPackets[len(d.recoveredPackets)-1]
+		if backRecoveredPacket.SSRC == pkt.SSRC {
+			if seqDiff(pkt.SequenceNumber, backRecoveredPacket.SequenceNumber) > uint16(d.maxMediaPackets) {
+				d.recoveredPackets = nil
+				d.receivedFECPackets = nil
+			}
+		}
+	}
+
+	d.insertPacket(pkt)
+
+	return d.attemptRecovery()
+}
+
+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 keeps the slice sorted
+	// and reduces incorrect decoding due to sequence number wrap-around.
+	if len(d.receivedFECPackets) > 0 && receivedPkt.SSRC == d.ssrc {
+		toRemove := 0
+		for _, fecPkt := range d.receivedFECPackets {
+			if abs(int(receivedPkt.SequenceNumber)-int(fecPkt.packet.SequenceNumber)) > 0x3fff {
+				toRemove++
+			} else {
+				break
+			}
+		}
+		if toRemove > 0 {
+			d.receivedFECPackets = d.receivedFECPackets[toRemove:]
+		}
+	}
+
+	switch receivedPkt.SSRC {
+	case d.ssrc:
+		d.insertFECPacket(receivedPkt)
+	case d.protectedStreamSSRC:
+		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) {
+	pkt := receivedPkt
+	for _, fecPkt := range d.receivedFECPackets {
+		for _, protectedPacket := range fecPkt.protectedPackets {
+			if protectedPacket.seq == pkt.SequenceNumber {
+				protectedPacket.packet = &pkt
+			}
+		}
+	}
+}
+
+func (d *Decoder) insertFECPacket(fecPkt rtp.Packet) {
+	for _, existingFECPacket := range d.receivedFECPackets {
+		if existingFECPacket.packet.SequenceNumber == fecPkt.SequenceNumber {
+			return
+		}
+	}
+
+	fec, err := parseFlexFEC03Header(fecPkt.Payload)
+	if err != nil {
+		return
+	}
+
+	if fec.protectedSSRC != d.protectedStreamSSRC {
+		return
+	}
+
+	protectedSeqs := decodeMask(uint64(fec.mask0), 15, fec.seqNumBase)
+	if fec.mask1 != 0 {
+		protectedSeqs = append(protectedSeqs, decodeMask(uint64(fec.mask1), 31, fec.seqNumBase+15)...)
+	}
+	if fec.mask2 != 0 {
+		protectedSeqs = append(protectedSeqs, decodeMask(fec.mask2, 63, fec.seqNumBase+46)...)
+	}
+
+	if len(protectedSeqs) == 0 {
+		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) > d.maxFECPackets {
+		d.receivedFECPackets = d.receivedFECPackets[1:]
+	}
+}
+
+func (d *Decoder) attemptRecovery() []rtp.Packet {
+	recoveredPackets := make([]rtp.Packet, 0)
+	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 {
+				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
+
+	// extract the FEC bit string as the first 80 bits of the FEC header.
+	headerRecovery := make([]byte, 12)
+	if len(fec.packet.Payload) < 10 {
+		return rtp.Packet{}, errPacketTruncated
+	}
+	copy(headerRecovery, fec.packet.Payload[:10])
+
+	var seqnum uint16
+	for _, protectedPacket := range fec.protectedPackets {
+		if protectedPacket.packet != nil {
+			// for each received source packet, compute the 80-bit string by
+			// concatenating the first 64 bits of its RTP header and the 16-bit
+			// network-ordered representation of its length in bytes minus 12.
+			receivedHeader, err := protectedPacket.packet.Header.Marshal()
+			if err != nil {
+				return rtp.Packet{}, fmt.Errorf("marshal received header: %w", err)
+			}
+			binary.BigEndian.PutUint16(receivedHeader[2:4], uint16(protectedPacket.packet.MarshalSize()-12))
+			for i := range 8 {
+				headerRecovery[i] ^= receivedHeader[i]
+			}
+		} else {
+			seqnum = protectedPacket.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.protectedStreamSSRC)
+
+	payloadRecovery := make([]byte, payloadLength)
+	copy(payloadRecovery, fec.flexFec.payload)
+	for _, protectedPacket := range fec.protectedPackets {
+		if protectedPacket.packet != nil {
+			packet, err := protectedPacket.packet.Marshal()
+			if err != nil {
+				return rtp.Packet{}, 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...)
+
+	var packet rtp.Packet
+	if err := packet.Unmarshal(headerRecovery); err != nil {
+		return rtp.Packet{}, fmt.Errorf("unmarshal recovered: %w", err)
+	}
+
+	return packet, nil
+}
+
+func (d *Decoder) discardOldRecoveredPackets() {
+	if len(d.recoveredPackets) > recoveredPacketLimit {
+		d.recoveredPackets = d.recoveredPackets[len(d.recoveredPackets)-recoveredPacketLimit:]
+	}
+}
+
+func decodeMask(mask uint64, bitCount uint16, seqNumBase uint16) []uint16 {
+	res := make([]uint16, 0)
+	for i := uint16(0); i < bitCount; i++ {
+		if (mask>>(bitCount-1-i))&1 == 1 {
+			res = append(res, seqNumBase+i)
+		}
+	}
+
+	return res
+}
+
+type fecPacketState struct {
+	packet           rtp.Packet
+	flexFec          flexFec
+	protectedPackets []*protectedPacket
+}
+
+type flexFec struct {
+	protectedSSRC uint32
+	seqNumBase    uint16
+	mask0         uint16
+	mask1         uint32
+	mask2         uint64
+	payload       []byte
+}
+
+type protectedPacket struct {
+	seq    uint16
+	packet *rtp.Packet
+}
+
+func parseFlexFEC03Header(data []byte) (flexFec, error) {
+	if len(data) < 20 {
+		return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data))
+	}
+
+	rBit := (data[0] & 0x80) != 0
+	if rBit {
+		return flexFec{}, errRetransmissionBitSet
+	}
+
+	fBit := (data[0] & 0x40) != 0
+	if fBit {
+		return flexFec{}, errInflexibleGeneratorMatrix
+	}
+
+	ssrcCount := data[8]
+	if ssrcCount != 1 {
+		return flexFec{}, fmt.Errorf("%w: count %d", errMultipleSSRCProtection, ssrcCount)
+	}
+
+	protectedSSRC := binary.BigEndian.Uint32(data[12:])
+	seqNumBase := binary.BigEndian.Uint16(data[16:])
+	rawPacketMask := data[18:]
+	var payload []byte
+
+	kBit0 := (rawPacketMask[0] & 0x80) != 0
+	maskPart0 := binary.BigEndian.Uint16(rawPacketMask[0:2]) & 0x7FFF
+	var maskPart1 uint32
+	var maskPart2 uint64
+
+	if kBit0 {
+		payload = rawPacketMask[2:]
+	} else {
+		if len(data) < 24 {
+			return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data))
+		}
+
+		kBit1 := (rawPacketMask[2] & 0x80) != 0
+		maskPart1 = binary.BigEndian.Uint32(rawPacketMask[2:]) & 0x7FFFFFFF
+
+		if kBit1 {
+			payload = rawPacketMask[6:]
+		} else {
+			if len(data) < 32 {
+				return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data))
+			}
+
+			kBit2 := (rawPacketMask[6] & 0x80) != 0
+			maskPart2 = binary.BigEndian.Uint64(rawPacketMask[6:]) & 0x7FFFFFFFFFFFFFFF
+
+			if kBit2 {
+				payload = rawPacketMask[14:]
+			} else {
+				return flexFec{}, errLastOptionalMaskKBitSetToFalse
+			}
+		}
+	}
+
+	return flexFec{
+		protectedSSRC: protectedSSRC,
+		seqNumBase:    seqNumBase,
+		mask0:         maskPart0,
+		mask1:         maskPart1,
+		mask2:         maskPart2,
+		payload:       payload,
+	}, nil
+}
+
+func clonePacket(pkt rtp.Packet) rtp.Packet {
+	cloned := pkt
+	cloned.Header = pkt.Header.Clone()
+	if pkt.Payload != nil {
+		cloned.Payload = make([]byte, len(pkt.Payload))
+		copy(cloned.Payload, pkt.Payload)
+	}
+	return cloned
+}
+
+func seqDiff(a, b uint16) uint16 {
+	return min(a-b, b-a)
+}
+
+func abs(x int) int {
+	if x >= 0 {
+		return x
+	}
+
+	return -x
+}
+
+func isNewerSeq(prevValue, value uint16) bool {
+	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,173 @@
+// Copyright 2024 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 (
+	"testing"
+
+	"github.com/pion/interceptor/pkg/flexfec"
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	testMediaSSRC   = uint32(0x11223344)
+	testFECSSRC     = uint32(0x55667788)
+	testFECPayloadT = uint8(49)
+	testMediaPT     = uint8(96)
+)
+
+func makeMediaPackets(baseSeq uint16, count int) []rtp.Packet {
+	pkts := make([]rtp.Packet, 0, count)
+	for i := 0; i < count; i++ {
+		// vary the payload length to exercise length recovery
+		payloadLen := 20 + (i % 5)
+		payload := make([]byte, payloadLen)
+		for j := range payload {
+			payload[j] = byte((i*7 + j*3) & 0xff)
+		}
+		pkts = append(pkts, rtp.Packet{
+			Header: rtp.Header{
+				Version:        2,
+				PayloadType:    testMediaPT,
+				SequenceNumber: baseSeq + uint16(i),
+				Timestamp:      uint32(1000 + i*960),
+				SSRC:           testMediaSSRC,
+			},
+			Payload: payload,
+		})
+	}
+	return pkts
+}
+
+func requirePacketsEqual(t *testing.T, want, got rtp.Packet) {
+	t.Helper()
+	require.Equal(t, want.SequenceNumber, got.SequenceNumber, "sequence number")
+	require.Equal(t, want.SSRC, got.SSRC, "ssrc")
+	require.Equal(t, want.PayloadType, got.PayloadType, "payload type")
+	require.Equal(t, want.Timestamp, got.Timestamp, "timestamp")
+	require.Equal(t, want.Payload, got.Payload, "payload")
+}
+
+func TestDecoderRecoversSingleLossMediaThenFEC(t *testing.T) {
+	media := makeMediaPackets(1000, 10)
+
+	encoder := flexfec.NewFlexEncoder03(testFECPayloadT, testFECSSRC)
+	fecPackets := encoder.EncodeFec(cloneAll(media), 1)
+	require.NotEmpty(t, fecPackets, "encoder should produce a FEC packet")
+
+	dropped := 4
+	dec := NewDecoder(testFECSSRC, testMediaSSRC)
+
+	// deliver all media packets except the dropped one
+	for i, p := range media {
+		if i == dropped {
+			continue
+		}
+		require.Empty(t, dec.Decode(p), "no recovery expected while feeding media")
+	}
+
+	// deliver the FEC packet - now exactly one protected packet is missing
+	var recovered []rtp.Packet
+	for _, fp := range fecPackets {
+		recovered = append(recovered, dec.Decode(fp)...)
+	}
+
+	require.Len(t, recovered, 1)
+	requirePacketsEqual(t, media[dropped], recovered[0])
+}
+
+func TestDecoderRecoversWhenFECArrivesFirst(t *testing.T) {
+	media := makeMediaPackets(2000, 8)
+
+	encoder := flexfec.NewFlexEncoder03(testFECPayloadT, testFECSSRC)
+	fecPackets := encoder.EncodeFec(cloneAll(media), 1)
+	require.NotEmpty(t, fecPackets)
+
+	dropped := 2
+	dec := NewDecoder(testFECSSRC, testMediaSSRC)
+
+	// FEC arrives before any media: nothing can be recovered yet
+	for _, fp := range fecPackets {
+		require.Empty(t, dec.Decode(fp))
+	}
+
+	var recovered []rtp.Packet
+	for i, p := range media {
+		if i == dropped {
+			continue
+		}
+		recovered = append(recovered, dec.Decode(p)...)
+	}
+
+	require.Len(t, recovered, 1)
+	requirePacketsEqual(t, media[dropped], recovered[0])
+}
+
+func TestDecoderNoLossNoRecovery(t *testing.T) {
+	media := makeMediaPackets(3000, 6)
+
+	encoder := flexfec.NewFlexEncoder03(testFECPayloadT, testFECSSRC)
+	fecPackets := encoder.EncodeFec(cloneAll(media), 1)
+	require.NotEmpty(t, fecPackets)
+
+	dec := NewDecoder(testFECSSRC, testMediaSSRC)
+	for _, p := range media {
+		require.Empty(t, dec.Decode(p))
+	}
+	for _, fp := range fecPackets {
+		require.Empty(t, dec.Decode(fp), "no recovery expected when nothing is lost")
+	}
+}
+
+func TestDecoderCannotRecoverDoubleLoss(t *testing.T) {
+	media := makeMediaPackets(4000, 10)
+
+	encoder := flexfec.NewFlexEncoder03(testFECPayloadT, testFECSSRC)
+	fecPackets := encoder.EncodeFec(cloneAll(media), 1)
+	require.NotEmpty(t, fecPackets)
+
+	dec := NewDecoder(testFECSSRC, testMediaSSRC)
+	// drop two packets covered by a single FEC packet -> unrecoverable
+	for i, p := range media {
+		if i == 3 || i == 6 {
+			continue
+		}
+		dec.Decode(p)
+	}
+
+	var recovered []rtp.Packet
+	for _, fp := range fecPackets {
+		recovered = append(recovered, dec.Decode(fp)...)
+	}
+	require.Empty(t, recovered, "single FEC packet cannot recover two losses")
+}
+
+func TestDecoderIgnoresUnrelatedSSRC(t *testing.T) {
+	dec := NewDecoder(testFECSSRC, testMediaSSRC)
+	pkt := rtp.Packet{
+		Header:  rtp.Header{Version: 2, SSRC: 0xdeadbeef, SequenceNumber: 1},
+		Payload: []byte{1, 2, 3},
+	}
+	require.Empty(t, dec.Decode(pkt))
+}
+
+func cloneAll(pkts []rtp.Packet) []rtp.Packet {
+	out := make([]rtp.Packet, len(pkts))
+	for i, p := range pkts {
+		out[i] = clonePacket(p)
+	}
+	return out
+}

pkg/sfu/flexfec/encoder.go

@@ -0,0 +1,142 @@
+// Copyright 2024 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 (
+	"sync"
+
+	"github.com/pion/interceptor"
+	pionflexfec "github.com/pion/interceptor/pkg/flexfec"
+	"github.com/pion/rtp"
+)
+
+// EncoderInterceptorFactory builds EncoderInterceptors with the configured block size.
+type EncoderInterceptorFactory struct {
+	numMediaPackets uint32
+	numFecPackets   uint32
+}
+
+// NewEncoderInterceptorFactory returns a factory for a FlexFEC-03 encoder interceptor
+// that generates numFecPackets repair packets for every numMediaPackets media packets.
+//
+// It exists because pion's own flexfec.FecInterceptor is unsafe to use with LiveKit's
+// pooled RTP payloads. That interceptor retains each media packet's payload slice *by
+// reference* and only XORs the bytes once a full block has accumulated. LiveKit, however,
+// returns the underlying payload buffer to a sync.Pool as soon as the packet has been
+// written downstream (see pacer.Base.SendPacket -> Pool.Put). By the time pion computes
+// the parity over a block, the earlier packets' buffers have already been recycled and
+// overwritten, so the generated FEC protects garbage. Receivers that lose a protected
+// packet then "recover" corrupt bytes and inject them into the media stream, causing
+// frame corruption and PLI storms even at very low loss rates.
+//
+// This interceptor is behaviourally identical to pion's, except that it copies the
+// payload when buffering so the parity is computed over the bytes that were actually
+// sent. The copy is bounded by numMediaPackets per stream.
+func NewEncoderInterceptorFactory(numMediaPackets, numFecPackets uint32) *EncoderInterceptorFactory {
+	return &EncoderInterceptorFactory{
+		numMediaPackets: numMediaPackets,
+		numFecPackets:   numFecPackets,
+	}
+}
+
+// NewInterceptor constructs a new EncoderInterceptor.
+func (f *EncoderInterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor, error) {
+	return &EncoderInterceptor{
+		streams:         make(map[uint32]*encoderStream),
+		numMediaPackets: f.numMediaPackets,
+		numFecPackets:   f.numFecPackets,
+	}, nil
+}
+
+type encoderStream struct {
+	mu           sync.Mutex
+	encoder      pionflexfec.FlexEncoder
+	packetBuffer []rtp.Packet
+}
+
+// EncoderInterceptor generates a FlexFEC-03 repair stream for outgoing media.
+type EncoderInterceptor struct {
+	interceptor.NoOp
+	mu              sync.Mutex
+	streams         map[uint32]*encoderStream
+	numMediaPackets uint32
+	numFecPackets   uint32
+}
+
+// UnbindLocalStream removes per-stream encoder state.
+func (e *EncoderInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) {
+	e.mu.Lock()
+	delete(e.streams, info.SSRC)
+	e.mu.Unlock()
+}
+
+// BindLocalStream wraps the writer so FEC repair packets are emitted alongside media.
+func (e *EncoderInterceptor) BindLocalStream(
+	info *interceptor.StreamInfo, writer interceptor.RTPWriter,
+) interceptor.RTPWriter {
+	// No FEC negotiated for this stream; pass through untouched.
+	if info.PayloadTypeForwardErrorCorrection == 0 || info.SSRCForwardErrorCorrection == 0 {
+		return writer
+	}
+
+	mediaSSRC := info.SSRC
+
+	stream := &encoderStream{
+		encoder: pionflexfec.FlexEncoder03Factory{}.NewEncoder(
+			info.PayloadTypeForwardErrorCorrection,
+			info.SSRCForwardErrorCorrection,
+		),
+	}
+
+	e.mu.Lock()
+	e.streams[mediaSSRC] = stream
+	e.mu.Unlock()
+
+	return interceptor.RTPWriterFunc(
+		func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
+			// Only the protected media stream feeds the encoder.
+			if header.SSRC != mediaSSRC {
+				return writer.Write(header, payload, attributes)
+			}
+
+			var fecPackets []rtp.Packet
+			stream.mu.Lock()
+			// Copy the payload: LiveKit hands us a pooled buffer that is recycled the
+			// moment this write returns, so it cannot be retained by reference.
+			payloadCopy := make([]byte, len(payload))
+			copy(payloadCopy, payload)
+			stream.packetBuffer = append(stream.packetBuffer, rtp.Packet{
+				Header:  *header,
+				Payload: payloadCopy,
+			})
+			if len(stream.packetBuffer) == int(e.numMediaPackets) {
+				fecPackets = stream.encoder.EncodeFec(stream.packetBuffer, e.numFecPackets)
+				stream.packetBuffer = nil
+			}
+			stream.mu.Unlock()
+
+			result, err := writer.Write(header, payload, attributes)
+
+			for i := range fecPackets {
+				fecHeader := fecPackets[i].Header
+				if _, fecErr := writer.Write(&fecHeader, fecPackets[i].Payload, attributes); fecErr != nil && err == nil {
+					err = fecErr
+				}
+			}
+
+			return result, err
+		},
+	)
+}

pkg/sfu/flexfec/encoder_test.go

@@ -0,0 +1,160 @@
+// Copyright 2024 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 (
+	"testing"
+
+	"github.com/pion/interceptor"
+	pionflexfec "github.com/pion/interceptor/pkg/flexfec"
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/require"
+)
+
+// recyclingWriter mimics LiveKit's pooled-payload lifecycle: it captures a deep copy of
+// every packet it is asked to write (the bytes that are actually sent on the wire), then
+// overwrites the caller-supplied payload buffer to emulate pacer.Base.SendPacket
+// returning that buffer to the sync.Pool. Any encoder that retained the payload slice by
+// reference will subsequently XOR garbage.
+type recyclingWriter struct {
+	fecSSRC uint32
+	media   []rtp.Packet
+	fec     []rtp.Packet
+}
+
+func (w *recyclingWriter) Write(header *rtp.Header, payload []byte, _ interceptor.Attributes) (int, error) {
+	captured := make([]byte, len(payload))
+	copy(captured, payload)
+	pkt := rtp.Packet{Header: *header, Payload: captured}
+	if header.SSRC == w.fecSSRC {
+		w.fec = append(w.fec, pkt)
+	} else {
+		w.media = append(w.media, pkt)
+	}
+
+	// Recycle the caller's buffer, as the pacer does once the write returns.
+	for i := range payload {
+		payload[i] = 0xff
+	}
+	return len(payload), nil
+}
+
+func fecStreamInfo() *interceptor.StreamInfo {
+	return &interceptor.StreamInfo{
+		SSRC:                              testMediaSSRC,
+		SSRCForwardErrorCorrection:        testFECSSRC,
+		PayloadTypeForwardErrorCorrection: testFECPayloadT,
+	}
+}
+
+// drivePooled writes count media packets through wr. Each packet is backed by its own
+// freshly allocated buffer (standing in for a pooled buffer) so that recyclingWriter can
+// safely scribble over it after each write.
+func drivePooled(wr interceptor.RTPWriter, baseSeq uint16, count int) {
+	for i := 0; i < count; i++ {
+		payloadLen := 20 + (i % 5)
+		buf := make([]byte, payloadLen)
+		for j := range buf {
+			buf[j] = byte((i*7 + j*3) & 0xff)
+		}
+		hdr := &rtp.Header{
+			Version:        2,
+			PayloadType:    testMediaPT,
+			SequenceNumber: baseSeq + uint16(i),
+			Timestamp:      uint32(1000 + i*960),
+			SSRC:           testMediaSSRC,
+		}
+		_, _ = wr.Write(hdr, buf, nil)
+	}
+}
+
+// TestEncoderInterceptorRecoversWithPooledPayloads is the regression test for the FEC
+// generation bug: our interceptor must copy the payload before retaining it, so that the
+// parity it computes matches the bytes actually sent even though the caller recycles the
+// payload buffer immediately after each write.
+func TestEncoderInterceptorRecoversWithPooledPayloads(t *testing.T) {
+	const numMedia, numFec = 5, uint32(1)
+
+	wr := &recyclingWriter{fecSSRC: testFECSSRC}
+	factory := NewEncoderInterceptorFactory(numMedia, numFec)
+	itc, err := factory.NewInterceptor("")
+	require.NoError(t, err)
+	writer := itc.BindLocalStream(fecStreamInfo(), interceptor.RTPWriterFunc(wr.Write))
+
+	drivePooled(writer, 1000, numMedia)
+
+	require.Len(t, wr.media, numMedia, "all media packets forwarded")
+	require.Len(t, wr.fec, int(numFec), "one FEC packet generated for the block")
+
+	// Drop one media packet, deliver the rest plus FEC, and confirm the recovered
+	// packet exactly matches the bytes that were sent.
+	const dropped = 2
+	dec := NewDecoder(testFECSSRC, testMediaSSRC)
+	for i, p := range wr.media {
+		if i == dropped {
+			continue
+		}
+		require.Empty(t, dec.Decode(p))
+	}
+
+	var recovered []rtp.Packet
+	for _, fp := range wr.fec {
+		recovered = append(recovered, dec.Decode(fp)...)
+	}
+
+	require.Len(t, recovered, 1, "FEC must recover the single dropped packet")
+	requirePacketsEqual(t, wr.media[dropped], recovered[0])
+}
+
+// TestPionEncoderCorruptsWithPooledPayloads documents the upstream behaviour that makes
+// pion's own encoder interceptor unsafe with LiveKit's pooled payloads: because it
+// retains the payload by reference, recycling the buffer corrupts the parity and the
+// "recovered" packet does not match what was sent. This is the bug our interceptor fixes.
+func TestPionEncoderCorruptsWithPooledPayloads(t *testing.T) {
+	const numMedia, numFec = 5, uint32(1)
+
+	wr := &recyclingWriter{fecSSRC: testFECSSRC}
+	factory, err := pionflexfec.NewFecInterceptor(
+		pionflexfec.NumMediaPackets(numMedia),
+		pionflexfec.NumFECPackets(numFec),
+	)
+	require.NoError(t, err)
+	itc, err := factory.NewInterceptor("")
+	require.NoError(t, err)
+	writer := itc.BindLocalStream(fecStreamInfo(), interceptor.RTPWriterFunc(wr.Write))
+
+	drivePooled(writer, 1000, numMedia)
+	require.Len(t, wr.media, numMedia)
+	require.Len(t, wr.fec, int(numFec))
+
+	const dropped = 2
+	dec := NewDecoder(testFECSSRC, testMediaSSRC)
+	for i, p := range wr.media {
+		if i == dropped {
+			continue
+		}
+		dec.Decode(p)
+	}
+	var recovered []rtp.Packet
+	for _, fp := range wr.fec {
+		recovered = append(recovered, dec.Decode(fp)...)
+	}
+
+	// pion still "recovers" a packet, but its payload is garbage because the parity was
+	// computed over recycled buffers.
+	require.Len(t, recovered, 1)
+	require.NotEqual(t, wr.media[dropped].Payload, recovered[0].Payload,
+		"pion encoder is expected to produce corrupt recovery with pooled payloads")
+}

scripts/flexfec/README.md

@@ -0,0 +1,145 @@
+# Local FlexFEC-03 test harness
+
+Scripts to validate end-to-end FlexFEC-03 between the rust-sdks `local_video`
+examples and a locally-built SFU, under simulated packet loss.
+
+FlexFEC is terminated at the SFU (it can't be passed through because the SFU
+rewrites SSRC/sequence numbers per subscriber), so the two directions are
+independent and tested separately:
+
+| Direction | What's exercised                                   | Loss to inject |
+| --------- | -------------------------------------------------- | -------------- |
+| Downlink  | SFU **generates** a repair stream per subscriber   | `down`         |
+| Uplink    | SFU **receives + decodes** a publisher repair flow | `up`           |
+
+## Prerequisites
+
+- Go toolchain (to build this SFU).
+- A `rust-sdks5` checkout with the FlexFEC changes, next to this repo
+  (`../rust-sdks5`) or pointed at via `RUST_SDKS_DIR`. Its `webrtc-sys` must be
+  built (the C++ field trials that enable FlexFEC-03 live there).
+- A camera (the publisher captures real video).
+- `sudo` for traffic shaping. macOS uses `dnctl`/`pfctl` (dummynet); Linux uses
+  `tc netem`.
+
+All processes run on one machine, so media flows over loopback. The SFU is pinned
+to UDP `7882` (see `flexfec-local.yaml`) so shaping can target it by port.
+
+## Files
+
+| File                 | Purpose                                              |
+| -------------------- | ---------------------------------------------------- |
+| `flexfec-local.yaml` | SFU config: flexfec on (pub+sub), fixed loopback port |
+| `run-sfu.sh`         | Build + run the SFU (`--dev`, devkey/secret)         |
+| `netem.sh`           | Start/stop/inspect packet-loss shaping on udp/7882   |
+| `run-publisher.sh`   | Run the publisher with `--flex-fec`                  |
+| `run-subscriber.sh`  | Run the subscriber, surface `Video FEC ...` logs     |
+
+## Workflow
+
+Use four terminals.
+
+**1 — SFU**
+
+```bash
+scripts/flexfec/run-sfu.sh
+```
+
+Connects on `ws://127.0.0.1:7880` (API key `devkey` / secret `secret`).
+
+**2 — Subscriber**
+
+```bash
+scripts/flexfec/run-subscriber.sh
+# or only show FEC lines on the console:
+FEC_ONLY=1 scripts/flexfec/run-subscriber.sh
+```
+
+**3 — Publisher** (grant camera permission on first run)
+
+```bash
+scripts/flexfec/run-publisher.sh
+```
+
+**4 — Packet loss** (needs sudo)
+
+```bash
+# Uplink test: drop 8% of publisher -> SFU packets
+sudo scripts/flexfec/netem.sh start 8 up
+
+# Downlink test: drop 8% of SFU -> subscriber packets (+ optional delay_ms, see A/B below)
+sudo scripts/flexfec/netem.sh start 8 down
+sudo scripts/flexfec/netem.sh start 8 down 150   # 8% loss + 150ms one-way delay
+
+sudo scripts/flexfec/netem.sh status
+sudo scripts/flexfec/netem.sh stop   # always clean up when done
+```
+
+## What success looks like
+
+**Downlink (`down`)** — the subscriber recovers SFU-generated repair packets.
+In the subscriber log (`logs/subscriber.log`):
+
+```
+Video FEC active: fec_ssrc=..., received_total=..., discarded_total=..., accepted_recovery_payload_total=...
+Video FEC recovery payload accepted: +N packets (accepted_total=...)
+```
+
+`accepted_recovery_payload_total` should climb while loss is applied.
+
+**Uplink (`up`)** — the SFU receives + decodes the publisher's repair stream.
+In the SFU log:
+
+```
+flexfec decoder enabled   {"fecSSRC": ..., "mediaSSRC": ...}
+flexfec recovery stats    {"fecPacketsReceived": ..., "packetsRecovered": ..., "packetsRecoveredDelta": ...}
+```
+
+`packetsRecovered` increasing confirms the SFU rebuilt lost uplink packets.
+
+To also see FlexFEC SDP negotiation (`flexfec ... from sdp`, codec registration),
+set `logging.level: debug` in `flexfec-local.yaml`.
+
+### Subscriber-side A/B (isolating FEC from RTX)
+
+On loopback RTT~=0, so NACK/RTX recovers nearly all loss before FlexFEC matters and
+its benefit is invisible. To make FEC the primary recovery path, add one-way delay so
+retransmissions arrive too late, and compare FEC on vs off under the same conditions:
+
+```bash
+# A) FEC ON (flexfec.subscriber: true). With SFU + pub + sub running:
+sudo scripts/flexfec/netem.sh start 8 down 150     # 8% loss + 150ms delay, downlink
+# ...observe subscriber for ~20s, then...
+sudo scripts/flexfec/netem.sh stop
+
+# B) FEC OFF: set flexfec.subscriber: false in flexfec-local.yaml, restart the SFU,
+#    reconnect the subscriber, then apply the SAME shaping and observe.
+```
+
+The subscriber logs a `Video quality:` line every ~3s:
+
+```
+Video quality: frames_decoded=..., frames_dropped=..., freezes=N (Xs), nack=..., pli=...,
+               packets_lost=..., rtx_recovered=..., fec_recv=..., fec_discarded=...
+```
+
+With FEC ON you expect fewer `freezes`/`frames_dropped`, lower net `packets_lost`, and a
+meaningful `fec_recv`; with FEC OFF the same loss leans entirely on `rtx_recovered` and
+(because of the delay) typically shows more freezes and dropped frames.
+
+## Notes & caveats
+
+- **Loopback shaping (macOS):** `netem.sh` loads its dummynet rules into a dedicated
+  `flexfec-loss` pf anchor and re-applies `/etc/pf.conf` non-destructively. `stop`
+  flushes the pipes and restores pf to its prior enabled/disabled state. If pf is set
+  to `skip on lo0` the script warns and shaping won't apply.
+- **Direction isolation is approximate.** Both clients share the SFU UDP port, so
+  `up`/`down` are distinguished by source vs. destination port. The publisher is the
+  dominant sender to the SFU and the subscriber the dominant receiver, so this is good
+  enough to isolate each path in practice (some RTCP is also affected).
+- **`dnctl -q flush` clears all dummynet pipes**, not just ours — fine on a dev box.
+- **Send-side BWE:** if the SFU uses pacer-based send-side BWE, subscriber FEC overhead
+  isn't accounted for by congestion control (logged as a warning by the SFU). The
+  interceptor-based send-side BWE path TWCC-tags FEC packets correctly.
+- Increase loss gradually (e.g. 3% → 10%). Beyond what a single FlexFEC repair packet
+  can cover, bursts become unrecoverable (expected).

scripts/flexfec/flexfec-linux.yaml

@@ -0,0 +1,45 @@
+# Linux FlexFEC-03 test config for the netns + netem harness (scripts/flexfec/netns.sh).
+#
+# Unlike flexfec-local.yaml (macOS, loopback + in-SFU software impairment), this runs the
+# SFU in the root netns and advertises the veth host IP 10.200.0.1 so that:
+#   - the publisher in the `robot` netns reaches the SFU over the veth pair (shaped by netem)
+#   - the subscriber in the root netns reaches the SFU at 10.200.0.1 locally (clean link)
+#
+# Run with --dev (injects devkey/secret):
+#   ~/go1.26/bin/... # build first, then:
+#   sudo scripts/flexfec/netns.sh up
+#   /tmp/flexfec-livekit-server --dev --config scripts/flexfec/flexfec-linux.yaml
+
+port: 7880
+
+# Bind signalling (HTTP/WS + TCP) to the veth host IP so the publisher in the `robot` netns
+# can reach it, plus loopback for the in-root-ns subscriber. Setting a non-loopback bind
+# address also stops --dev from pinning everything to 127.0.0.1.
+bind_addresses:
+  - 10.200.0.1
+  - 127.0.0.1
+
+rtc:
+  udp_port: 7882
+  tcp_port: 7881
+  # Advertise the veth host IP, not a public IP.
+  use_external_ip: false
+  enable_loopback_candidate: true
+  ips:
+    includes:
+      # veth subnet: 10.200.0.1 (SFU/root) <-> 10.200.0.2 (robot netns)
+      - 10.200.0.0/24
+      - 127.0.0.0/8
+
+  # Asymmetric teleop topology: robot publishes with FlexFEC (uplink recovery at the SFU);
+  # the operator/subscriber is on a clean wired link so downlink FEC is off.
+  flexfec:
+    subscriber: false
+    publisher: true
+    payload_type: 49
+    num_media_packets: 5
+    num_fec_packets: 1
+
+logging:
+  level: debug
+  json: false

scripts/flexfec/flexfec-local.yaml

@@ -0,0 +1,48 @@
+# Local FlexFEC-03 test config for the LiveKit SFU.
+#
+# Intended to be run together with `--dev` (which injects the devkey/secret API key
+# pair and binds to loopback):
+#
+#   go run ./cmd/server --dev --config scripts/flexfec/flexfec-local.yaml
+#
+# Media is pinned to a single UDP port on loopback so traffic shaping (see netem.sh)
+# can target it deterministically.
+
+port: 7880
+
+rtc:
+  # Single, fixed media port (no port_range_* so this takes effect). Shape this port
+  # with scripts/flexfec/netem.sh to simulate packet loss.
+  udp_port: 7882
+  tcp_port: 7881
+  # Keep everything on the local machine; do not try to discover a public IP.
+  use_external_ip: false
+  # Gather loopback (127.0.0.1) candidates and pin ICE to IPv4 loopback so all media flows
+  # over a single, stable path. NOTE: on an all-on-one-box macOS setup, pf/dummynet
+  # (netem.sh) cannot shape this traffic -- the kernel short-circuits same-host UDP delivery
+  # before the pf hook -- so loss/delay is injected in-SFU instead via the LK_DOWNLINK_*
+  # env vars (see pkg/sfu/downtrack_impair.go and run-sfu.sh). The media path is therefore
+  # irrelevant to shaping, so loopback (no dependency on a volatile LAN IP) is preferred.
+  enable_loopback_candidate: true
+  ips:
+    includes:
+      - 127.0.0.0/8
+
+  # FlexFEC-03 for both directions:
+  #   publisher:  receive + decode a publisher's repair stream (uplink recovery)
+  #   subscriber: generate a repair stream per subscriber (downlink recovery)
+  flexfec:
+    subscriber: false
+    publisher: true
+    payload_type: 49
+    # Low-latency teleop tuning: 5-packet block with 1 repair = 20% overhead, but the
+    # repair lands ~16ms after the block (vs ~32ms for a 10-block), well under an 80ms
+    # RTX round trip. Recovers 1 scattered loss per 5-block (ample for ~1% random loss).
+    num_media_packets: 5
+    num_fec_packets: 1
+
+logging:
+  # info surfaces the "flexfec decoder enabled" and periodic "flexfec recovery stats"
+  # logs. Bump to debug to also see SDP-level FEC negotiation ("flexfec ... from sdp").
+  level: debug
+  json: false

scripts/flexfec/netem.sh

@@ -0,0 +1,289 @@
+#!/usr/bin/env bash
+#
+# Simulate packet loss on the SFU's media UDP port so FlexFEC recovery can be
+# observed end-to-end. Works on macOS (dnctl/dummynet + pfctl) and Linux (tc netem).
+#
+# Because publisher, SFU and subscriber all run on one machine, media flows over the
+# loopback interface. Loss is matched by the SFU UDP port (default 7882):
+#   - "down" drops packets coming FROM the SFU port  -> SFU -> subscriber path
+#            (exercises subscriber-side / downlink FEC generation+recovery)
+#   - "up"   drops packets going TO the SFU port      -> publisher -> SFU path
+#            (exercises publisher-side / uplink FEC reception+recovery)
+#   - "both" drops in both directions
+#
+# Usage:
+#   sudo scripts/flexfec/netem.sh start [loss_percent] [up|down|both] [delay_ms]  # uniform loss; default: 5 down 0
+#   sudo scripts/flexfec/netem.sh burst [loss_percent] [burst_ms] [gap_ms] [up|down|both] [delay_ms]  # bursty loss
+#   sudo scripts/flexfec/netem.sh stop
+#   sudo scripts/flexfec/netem.sh status
+#
+# delay_ms adds one-way latency (so RTT ~= 2*delay_ms on loopback). Models the
+# robot's access-network round trip; RTX recovery costs ~1 RTT of buffering.
+#
+# "burst" emulates handoff/fade-style correlated loss: it toggles the loss rate
+# between loss_percent (for burst_ms) and 0 (for gap_ms) on a duty cycle, holding
+# the delay constant. Average loss ~= loss_percent * burst_ms/(burst_ms+gap_ms).
+# Run it in its own terminal; Ctrl-C stops and restores the network.
+# This is where RTX (retransmit storms, each costing an RTT) and FlexFEC (fixed
+# inline overhead) diverge most.
+#
+# Env overrides:
+#   SFU_UDP_PORT  media port to shape (default 7882)
+#   NETEM_IFACE   (Linux only) interface to shape (default lo)
+set -euo pipefail
+
+PORT="${SFU_UDP_PORT:-7882}"
+ANCHOR="flexfec-loss"
+STATE_DIR="${TMPDIR:-/tmp}/flexfec-netem"
+PF_STATE="$STATE_DIR/pf.enabled"
+OS="$(uname -s)"
+
+require_root() {
+  if [[ "$(id -u)" -ne 0 ]]; then
+    echo "error: this command needs root; re-run with sudo" >&2
+    exit 1
+  fi
+}
+
+validate_dir() {
+  case "$1" in
+    up | down | both) ;;
+    *)
+      echo "error: direction must be up|down|both (got '$1')" >&2
+      exit 1
+      ;;
+  esac
+}
+
+# ----------------------------------------------------------------------------- macOS
+
+# Install the pf anchor + dummynet rules for the requested direction. Pipes are
+# (re)configured separately so loss can be toggled live for burst mode.
+macos_setup() {
+  local dir="$1"
+
+  mkdir -p "$STATE_DIR"
+
+  # Remember whether pf was already enabled so `stop` can restore it.
+  if [[ ! -f "$PF_STATE" ]]; then
+    if pfctl -s info 2>/dev/null | grep -q 'Status: Enabled'; then
+      echo enabled >"$PF_STATE"
+    else
+      echo disabled >"$PF_STATE"
+    fi
+  fi
+
+  local rules=""
+  case "$dir" in
+    down | both) rules+="dummynet out proto udp from any port $PORT to any pipe 1"$'\n' ;;
+  esac
+  case "$dir" in
+    up | both) rules+="dummynet out proto udp from any to any port $PORT pipe 2"$'\n' ;;
+  esac
+
+  # Re-load the system ruleset plus references to our named anchor (non-destructive:
+  # existing /etc/pf.conf rules are preserved), then install our rules in the anchor.
+  {
+    cat /etc/pf.conf 2>/dev/null || true
+    echo "dummynet-anchor \"$ANCHOR\""
+    echo "anchor \"$ANCHOR\""
+  } | pfctl -f - 2>/dev/null
+  printf '%s' "$rules" | pfctl -a "$ANCHOR" -f - 2>/dev/null
+  pfctl -e 2>/dev/null || true
+
+  if pfctl -sa 2>/dev/null | grep -qi 'skip on lo'; then
+    echo "warning: pf is configured to 'skip on lo'; loopback shaping may not apply" >&2
+  fi
+}
+
+# macos_set_loss <loss_percent> <delay_ms>
+macos_set_loss() {
+  local plr
+  plr="$(awk "BEGIN { printf \"%.4f\", $1 / 100 }")"
+  dnctl pipe 1 config plr "$plr" delay "$2" >/dev/null
+  dnctl pipe 2 config plr "$plr" delay "$2" >/dev/null
+}
+
+macos_start() {
+  local loss="$1" dir="$2" delay="$3"
+  macos_setup "$dir"
+  macos_set_loss "$loss" "$delay"
+  echo "==> macOS dummynet loss enabled: ${loss}% loss, ${delay}ms delay (${dir}) on udp/${PORT}"
+}
+
+macos_burst() {
+  local loss="$1" burst_ms="$2" gap_ms="$3" dir="$4" delay="$5"
+  local burst_s gap_s avg
+  burst_s="$(awk "BEGIN { printf \"%.3f\", $burst_ms / 1000 }")"
+  gap_s="$(awk "BEGIN { printf \"%.3f\", $gap_ms / 1000 }")"
+  avg="$(awk "BEGIN { printf \"%.1f\", $loss * $burst_ms / ($burst_ms + $gap_ms) }")"
+
+  macos_setup "$dir"
+  macos_set_loss 0 "$delay"
+
+  trap 'echo; echo "==> stopping burst"; macos_stop; exit 0' INT TERM
+  echo "==> macOS bursty loss: ${loss}% for ${burst_ms}ms every $((burst_ms + gap_ms))ms (~${avg}% avg), ${delay}ms delay (${dir}) on udp/${PORT}"
+  echo "    Ctrl-C to stop and restore the network."
+  while true; do
+    macos_set_loss "$loss" "$delay"
+    printf '\r  [burst ON  %s%% loss]   ' "$loss"
+    sleep "$burst_s"
+    macos_set_loss 0 "$delay"
+    printf '\r  [burst off  0%% loss]   ' "$loss"
+    sleep "$gap_s"
+  done
+}
+
+macos_stop() {
+  dnctl -q flush 2>/dev/null || true
+  printf '' | pfctl -a "$ANCHOR" -f - 2>/dev/null || true
+  pfctl -f /etc/pf.conf 2>/dev/null || true
+  if [[ -f "$PF_STATE" && "$(cat "$PF_STATE")" == "disabled" ]]; then
+    pfctl -d 2>/dev/null || true
+  fi
+  rm -f "$PF_STATE"
+  echo "==> macOS dummynet loss disabled"
+}
+
+macos_status() {
+  echo "== dummynet pipes =="
+  dnctl list 2>/dev/null || true
+  echo "== pf anchor '$ANCHOR' =="
+  pfctl -a "$ANCHOR" -s rules 2>/dev/null || true
+  echo "== pf status =="
+  pfctl -s info 2>/dev/null | head -1 || true
+}
+
+# ----------------------------------------------------------------------------- Linux
+
+# linux_setup <dir> <delay_ms>: build prio qdisc + netem band (loss starts at 0)
+# and funnel matching UDP packets into it.
+linux_setup() {
+  local dir="$1" delay="$2"
+  local dev="${NETEM_IFACE:-lo}"
+
+  local netem_args=(loss 0%)
+  if [[ "$delay" -gt 0 ]]; then
+    netem_args+=(delay "${delay}ms")
+  fi
+
+  tc qdisc del dev "$dev" root 2>/dev/null || true
+  tc qdisc add dev "$dev" root handle 1: prio
+  tc qdisc add dev "$dev" parent 1:3 handle 30: netem "${netem_args[@]}"
+
+  if [[ "$dir" == "down" || "$dir" == "both" ]]; then
+    tc filter add dev "$dev" parent 1:0 protocol ip prio 3 u32 \
+      match ip protocol 17 0xff match ip sport "$PORT" 0xffff flowid 1:3
+  fi
+  if [[ "$dir" == "up" || "$dir" == "both" ]]; then
+    tc filter add dev "$dev" parent 1:0 protocol ip prio 3 u32 \
+      match ip protocol 17 0xff match ip dport "$PORT" 0xffff flowid 1:3
+  fi
+}
+
+# linux_set_loss <loss_percent> <delay_ms>
+linux_set_loss() {
+  local dev="${NETEM_IFACE:-lo}"
+  if [[ "$2" -gt 0 ]]; then
+    tc qdisc change dev "$dev" parent 1:3 handle 30: netem loss "${1}%" delay "${2}ms"
+  else
+    tc qdisc change dev "$dev" parent 1:3 handle 30: netem loss "${1}%"
+  fi
+}
+
+linux_start() {
+  local loss="$1" dir="$2" delay="$3"
+  local dev="${NETEM_IFACE:-lo}"
+  linux_setup "$dir" "$delay"
+  linux_set_loss "$loss" "$delay"
+  echo "==> Linux netem enabled: ${loss}% loss, ${delay}ms delay (${dir}) on udp/${PORT} dev ${dev}"
+}
+
+linux_burst() {
+  local loss="$1" burst_ms="$2" gap_ms="$3" dir="$4" delay="$5"
+  local dev="${NETEM_IFACE:-lo}"
+  local burst_s gap_s avg
+  burst_s="$(awk "BEGIN { printf \"%.3f\", $burst_ms / 1000 }")"
+  gap_s="$(awk "BEGIN { printf \"%.3f\", $gap_ms / 1000 }")"
+  avg="$(awk "BEGIN { printf \"%.1f\", $loss * $burst_ms / ($burst_ms + $gap_ms) }")"
+
+  linux_setup "$dir" "$delay"
+
+  trap 'echo; echo "==> stopping burst"; linux_stop; exit 0' INT TERM
+  echo "==> Linux bursty loss: ${loss}% for ${burst_ms}ms every $((burst_ms + gap_ms))ms (~${avg}% avg), ${delay}ms delay (${dir}) on udp/${PORT} dev ${dev}"
+  echo "    Ctrl-C to stop and restore the network."
+  while true; do
+    linux_set_loss "$loss" "$delay"
+    printf '\r  [burst ON  %s%% loss]   ' "$loss"
+    sleep "$burst_s"
+    linux_set_loss 0 "$delay"
+    printf '\r  [burst off  0%% loss]   ' "$loss"
+    sleep "$gap_s"
+  done
+}
+
+linux_stop() {
+  local dev="${NETEM_IFACE:-lo}"
+  tc qdisc del dev "$dev" root 2>/dev/null || true
+  echo "==> Linux netem loss disabled (dev ${dev})"
+}
+
+linux_status() {
+  local dev="${NETEM_IFACE:-lo}"
+  echo "== qdisc (dev ${dev}) =="
+  tc qdisc show dev "$dev" || true
+  echo "== filters (dev ${dev}) =="
+  tc filter show dev "$dev" || true
+}
+
+# ----------------------------------------------------------------------------- dispatch
+
+cmd="${1:-}"
+
+case "$cmd" in
+  start)
+    require_root
+    loss="${2:-5}"
+    dir="${3:-down}"
+    delay="${4:-0}"
+    validate_dir "$dir"
+    case "$OS" in
+      Darwin) macos_start "$loss" "$dir" "$delay" ;;
+      Linux) linux_start "$loss" "$dir" "$delay" ;;
+      *) echo "error: unsupported OS '$OS'" >&2; exit 1 ;;
+    esac
+    ;;
+  burst)
+    require_root
+    loss="${2:-40}"
+    burst_ms="${3:-200}"
+    gap_ms="${4:-2000}"
+    dir="${5:-down}"
+    delay="${6:-20}"
+    validate_dir "$dir"
+    case "$OS" in
+      Darwin) macos_burst "$loss" "$burst_ms" "$gap_ms" "$dir" "$delay" ;;
+      Linux) linux_burst "$loss" "$burst_ms" "$gap_ms" "$dir" "$delay" ;;
+      *) echo "error: unsupported OS '$OS'" >&2; exit 1 ;;
+    esac
+    ;;
+  stop)
+    require_root
+    case "$OS" in
+      Darwin) macos_stop ;;
+      Linux) linux_stop ;;
+    esac
+    ;;
+  status)
+    case "$OS" in
+      Darwin) macos_status ;;
+      Linux) linux_status ;;
+    esac
+    ;;
+  *)
+    echo "usage: sudo $0 {start [loss%] [up|down|both] [delay_ms]" >&2
+    echo "               | burst [loss%] [burst_ms] [gap_ms] [up|down|both] [delay_ms]" >&2
+    echo "               | stop | status}" >&2
+    exit 1
+    ;;
+esac

scripts/flexfec/netns.sh

@@ -0,0 +1,131 @@
+#!/usr/bin/env bash
+#
+# Linux network-namespace + netem harness for the asymmetric teleop FlexFEC test.
+#
+# Topology (models a robot on a lossy 5G uplink talking to a wired operator):
+#
+#     robot netns (publisher)            root netns (SFU + subscriber/operator)
+#     10.200.0.2  veth-robot  <=========>  veth-host 10.200.0.1   [SFU :7880/:7882]
+#                              netem here                          [subscriber -> 10.200.0.1]
+#
+# Only the robot<->SFU leg traverses the veth pair, so netem on the veth shapes the
+# robot uplink/downlink ONLY. The subscriber (operator) sits in the root netns and reaches
+# the SFU at 10.200.0.1 locally, so its link stays clean -- exactly the teleop case where
+# the robot is on 5G and the operator is on a stable hardline.
+#
+# Loss is applied to the robot->SFU (uplink) direction; delay is applied symmetrically so a
+# NACK->RTX recovery costs a full RTT (the latency publisher-side FlexFEC avoids by letting
+# the SFU repair the stream inline).
+#
+# Usage:
+#   sudo scripts/flexfec/netns.sh up                 # create ns + veth (no impairment)
+#   sudo scripts/flexfec/netns.sh shape <loss%> <owd_ms>   # apply/replace netem
+#   sudo scripts/flexfec/netns.sh clear              # remove netem (clean link)
+#   sudo scripts/flexfec/netns.sh status             # show qdiscs + addrs
+#   sudo scripts/flexfec/netns.sh down               # tear everything down
+#   scripts/flexfec/netns.sh exec <cmd...>           # run cmd inside robot netns (as $SUDO_USER)
+#
+# Example:
+#   sudo scripts/flexfec/netns.sh up
+#   sudo scripts/flexfec/netns.sh shape 3 20         # 3% uplink loss, 20ms each way (40ms RTT)
+set -euo pipefail
+
+NS=robot
+VETH_HOST=veth-host
+VETH_ROBOT=veth-robot
+HOST_IP=10.200.0.1
+ROBOT_IP=10.200.0.2
+PREFIX=24
+
+# Privileged commands are run via passwordless sudo for ip/tc only (see the sudoers drop-in
+# in the harness docs). The script itself does NOT need to run as root.
+if [[ "$(id -u)" == "0" ]]; then
+  SUDO=""
+else
+  SUDO="sudo -n"
+fi
+
+cmd_up() {
+  # Idempotent: tear down any prior instance first.
+  $SUDO ip netns del "$NS" 2>/dev/null || true
+  $SUDO ip link del "$VETH_HOST" 2>/dev/null || true
+
+  $SUDO ip netns add "$NS"
+  $SUDO ip link add "$VETH_HOST" type veth peer name "$VETH_ROBOT"
+  $SUDO ip link set "$VETH_ROBOT" netns "$NS"
+
+  $SUDO ip addr add "$HOST_IP/$PREFIX" dev "$VETH_HOST"
+  $SUDO ip link set "$VETH_HOST" up
+
+  $SUDO ip netns exec "$NS" ip addr add "$ROBOT_IP/$PREFIX" dev "$VETH_ROBOT"
+  $SUDO ip netns exec "$NS" ip link set "$VETH_ROBOT" up
+  $SUDO ip netns exec "$NS" ip link set lo up
+
+  echo "up: $NS netns ready"
+  echo "  root: $VETH_HOST $HOST_IP/$PREFIX   robot: $VETH_ROBOT $ROBOT_IP/$PREFIX"
+  echo "  SFU should advertise $HOST_IP (use flexfec-linux.yaml); publisher connects to ws://$HOST_IP:7880"
+}
+
+# shape <loss_percent> <one_way_delay_ms> [jitter_ms] [corr_pct]
+# When jitter_ms/corr_pct are given they feed netem's loss correlation + delay jitter to
+# approximate bursty (e.g. 5G handoff) conditions rather than uniform random loss.
+cmd_shape() {
+  local loss="${1:?usage: shape <loss%> <owd_ms> [jitter_ms] [loss_corr%]}"
+  local owd="${2:?usage: shape <loss%> <owd_ms> [jitter_ms] [loss_corr%]}"
+  local jitter="${3:-0}"
+  local corr="${4:-0}"
+
+  local delayspec="delay ${owd}ms"
+  [[ "$jitter" != "0" ]] && delayspec="delay ${owd}ms ${jitter}ms distribution normal"
+  local lossspec="loss ${loss}%"
+  [[ "$corr" != "0" ]] && lossspec="loss ${loss}% ${corr}%"
+
+  # Uplink (robot -> SFU): loss + delay. Applied on robot-side egress.
+  $SUDO ip netns exec "$NS" tc qdisc replace dev "$VETH_ROBOT" root netem $delayspec $lossspec
+  # Downlink (SFU -> robot, carries NACK/RTCP/PLI): delay only, no loss (operator link is
+  # clean, but the return path still costs propagation delay so RTX pays a full RTT).
+  $SUDO tc qdisc replace dev "$VETH_HOST" root netem delay "${owd}ms"
+
+  echo "shape: uplink ${lossspec} ${delayspec} ; downlink delay=${owd}ms (RTT≈$((owd*2))ms)"
+}
+
+cmd_clear() {
+  $SUDO ip netns exec "$NS" tc qdisc del dev "$VETH_ROBOT" root 2>/dev/null || true
+  $SUDO tc qdisc del dev "$VETH_HOST" root 2>/dev/null || true
+  echo "clear: netem removed (clean link)"
+}
+
+cmd_status() {
+  echo "=== root ns: $VETH_HOST ==="
+  $SUDO ip addr show "$VETH_HOST" 2>/dev/null | sed -n '2,4p' || echo "  (absent)"
+  $SUDO tc qdisc show dev "$VETH_HOST" 2>/dev/null || true
+  echo "=== robot ns: $VETH_ROBOT ==="
+  $SUDO ip netns exec "$NS" ip addr show "$VETH_ROBOT" 2>/dev/null | sed -n '2,4p' || echo "  (absent)"
+  $SUDO ip netns exec "$NS" tc qdisc show dev "$VETH_ROBOT" 2>/dev/null || true
+}
+
+cmd_down() {
+  $SUDO ip netns del "$NS" 2>/dev/null || true
+  $SUDO ip link del "$VETH_HOST" 2>/dev/null || true
+  echo "down: torn down"
+}
+
+# exec <cmd...> : run inside robot netns, dropping back to the invoking user. The outer
+# `sudo ip netns exec` runs as root; the inner `sudo -u $user` drops privileges (root -> user
+# needs no password), so the publisher runs as the normal user with its env intact.
+cmd_exec() {
+  local user="${SUDO_USER:-$USER}"
+  exec $SUDO ip netns exec "$NS" sudo -u "$user" --preserve-env=LIVEKIT_URL,LIVEKIT_API_KEY,LIVEKIT_API_SECRET,RUST_LOG,ROOM,PUB_IDENTITY,PATH,HOME,CARGO_HOME,RUSTUP_HOME "$@"
+}
+
+action="${1:-}"
+shift || true
+case "$action" in
+  up)     cmd_up "$@" ;;
+  shape)  cmd_shape "$@" ;;
+  clear)  cmd_clear "$@" ;;
+  status) cmd_status "$@" ;;
+  down)   cmd_down "$@" ;;
+  exec)   cmd_exec "$@" ;;
+  *) echo "usage: $0 {up|shape <loss%> <owd_ms>|clear|status|down|exec <cmd...>}" >&2; exit 2 ;;
+esac

scripts/flexfec/run-leg-linux.sh

@@ -0,0 +1,133 @@
+#!/usr/bin/env bash
+#
+# Run one FlexFEC test "leg" on the Linux box: start the SFU, a publisher (robot), and a
+# subscriber (operator), let them run, then print a summary of SFU FEC recovery + the
+# subscriber's frame-latency / video-quality stats.
+#
+# Two modes:
+#   - loopback (default): everything on 127.0.0.1, no impairment. Smoke test.
+#   - netns (USE_NETNS=1): publisher runs inside the `robot` netns (set up + shaped via
+#     netns.sh), SFU + subscriber run in the root netns reaching the SFU at 10.200.0.1.
+#     This models a lossy robot uplink with a clean operator downlink.
+#
+# Env:
+#   LABEL          leg label for the banner
+#   DURATION       seconds to run (default 35)
+#   USE_NETNS      1 to run publisher in the robot netns (default 0 = loopback)
+#   SFU_BIN        path to SFU binary (default /tmp/flexfec-livekit-server)
+#   RUST_DIR       rust-sdks5 dir (default ~/workspace/rust-sdks5)
+#   LK_DIR         livekit-flexfec dir (default ~/workspace/livekit-flexfec)
+#   PUB_EXTRA      extra publisher args (e.g. "--max-playout-delay 30")
+set -uo pipefail
+
+LABEL="${LABEL:-leg}"
+DURATION="${DURATION:-35}"
+USE_NETNS="${USE_NETNS:-0}"
+SFU_BIN="${SFU_BIN:-/tmp/flexfec-livekit-server}"
+RUST_DIR="${RUST_DIR:-$HOME/workspace/rust-sdks5}"
+LK_DIR="${LK_DIR:-$HOME/workspace/livekit-flexfec}"
+PUB_EXTRA="${PUB_EXTRA:-}"
+
+if [[ "$USE_NETNS" == "1" ]]; then
+  SFU_HOST="10.200.0.1"
+else
+  SFU_HOST="127.0.0.1"
+fi
+
+# FEC=on (default) uses flexfec-linux.yaml (publisher: true). FEC=off writes a sibling
+# config with publisher: false so the SFU does not negotiate uplink FlexFEC and the robot
+# falls back to RTX-only -- the A/B baseline.
+BASE_CONFIG="$LK_DIR/scripts/flexfec/flexfec-linux.yaml"
+if [[ "${FEC:-on}" == "off" ]]; then
+  CONFIG="/tmp/flexfec-linux-fecoff.yaml"
+  sed -E 's/^( *publisher:) *true/\1 false/' "$BASE_CONFIG" > "$CONFIG"
+else
+  CONFIG="$BASE_CONFIG"
+fi
+
+LOGDIR=/tmp/flexfec-logs
+mkdir -p "$LOGDIR"
+SFU_LOG="$LOGDIR/sfu.log"
+PUB_LOG="$LOGDIR/pub.log"
+SUB_LOG="$LOGDIR/sub.log"
+
+cleanup() {
+  pkill -f flexfec-livekit-server 2>/dev/null
+  pkill -f "target/debug/publisher" 2>/dev/null
+  pkill -f "target/debug/subscriber" 2>/dev/null
+}
+cleanup
+sleep 1
+
+echo "==================== LEG: $LABEL (netns=$USE_NETNS, ${DURATION}s) ===================="
+
+# In netns mode, apply netem shaping to the robot uplink. NETEM_LOSS/% NETEM_OWD ms, plus
+# optional NETEM_JITTER ms and NETEM_CORR % for bursty (5G-like) loss/jitter.
+if [[ "$USE_NETNS" == "1" ]]; then
+  NETEM_LOSS="${NETEM_LOSS:-0}"
+  NETEM_OWD="${NETEM_OWD:-0}"
+  if [[ "$NETEM_LOSS" != "0" || "$NETEM_OWD" != "0" ]]; then
+    "$LK_DIR/scripts/flexfec/netns.sh" shape "$NETEM_LOSS" "$NETEM_OWD" "${NETEM_JITTER:-0}" "${NETEM_CORR:-0}"
+  else
+    "$LK_DIR/scripts/flexfec/netns.sh" clear
+  fi
+fi
+
+# --- SFU (root netns) ---
+# In-SFU software impairment (used when netem/root is unavailable). All no-ops unless set.
+#   PUB_LOSS / PUB_DELAY_MS        robot uplink (publisher->SFU) loss + one-way delay
+#   DOWNLINK_LOSS / DOWNLINK_DELAY_MS / UPLINK_DELAY_MS   operator downlink + NACK delay
+( cd "$LK_DIR" && \
+  LK_PUB_LOSS="${PUB_LOSS:-0}" LK_PUB_DELAY_MS="${PUB_DELAY_MS:-0}" \
+  LK_DOWNLINK_LOSS="${DOWNLINK_LOSS:-0}" LK_DOWNLINK_DELAY_MS="${DOWNLINK_DELAY_MS:-0}" \
+  LK_UPLINK_DELAY_MS="${UPLINK_DELAY_MS:-0}" \
+  "$SFU_BIN" --dev --config "$CONFIG" > "$SFU_LOG" 2>&1 ) &
+sleep 3
+
+export LIVEKIT_URL="ws://$SFU_HOST:7880"
+export LIVEKIT_API_KEY=devkey
+export LIVEKIT_API_SECRET=secret
+export RUST_LOG="${RUST_LOG:-info}"
+ROOM=flexfec-test
+
+PUB="$RUST_DIR/target/debug/publisher"
+SUB="$RUST_DIR/target/debug/subscriber"
+
+# --- Publisher (robot) ---
+PUB_ARGS=(--flex-fec --room-name "$ROOM" --identity flexfec-pub --test-pattern --animate-test-pattern --attach-timestamp --attach-frame-id)
+# shellcheck disable=SC2206
+[[ -n "$PUB_EXTRA" ]] && PUB_ARGS+=($PUB_EXTRA)
+
+if [[ "$USE_NETNS" == "1" ]]; then
+  # netns.sh sudoes ip/tc internally; do NOT wrap it in sudo (the script isn't allowlisted).
+  "$LK_DIR/scripts/flexfec/netns.sh" exec env \
+    LIVEKIT_URL="$LIVEKIT_URL" LIVEKIT_API_KEY=devkey LIVEKIT_API_SECRET=secret RUST_LOG="$RUST_LOG" \
+    "$PUB" "${PUB_ARGS[@]}" > "$PUB_LOG" 2>&1 &
+else
+  "$PUB" "${PUB_ARGS[@]}" > "$PUB_LOG" 2>&1 &
+fi
+sleep 4
+
+# --- Subscriber (operator, root netns) ---
+# The subscriber opens an eframe/wgpu window (frame-latency stats are produced by its render
+# loop), so it needs a display. The box has a real X server on :1; use it unless DISPLAY is
+# already set. xvfb-run is used as a fallback if available.
+SUB_DISPLAY="${DISPLAY:-:1}"
+SUB_XAUTH="${XAUTHORITY:-/run/user/$(id -u)/gdm/Xauthority}"
+DISPLAY="$SUB_DISPLAY" XAUTHORITY="$SUB_XAUTH" "$SUB" --room-name "$ROOM" --identity flexfec-sub > "$SUB_LOG" 2>&1 &
+
+sleep "$DURATION"
+cleanup
+sleep 1
+
+echo "-- SFU FlexFEC recovery (publisher uplink), last 3 --"
+grep -iE 'flexfec recovery stats' "$SFU_LOG" 2>/dev/null | tail -3 | sed -E 's/.*"fecPacketsReceived"/fecRecv/; s/, "fecReportedReceived.*//'
+echo "-- SFU flexfec decoder enabled? --"
+grep -iE 'flexfec decoder enabled' "$SFU_LOG" 2>/dev/null | tail -1
+echo "-- frame latency, last 6 windows --"
+grep -iE 'frame latency' "$SUB_LOG" 2>/dev/null | tail -6
+echo "-- video quality (last) --"
+grep -iE 'Video quality' "$SUB_LOG" 2>/dev/null | tail -1
+echo "-- jitter buffer (last) --"
+grep -iE 'jitter buffer' "$SUB_LOG" 2>/dev/null | tail -1
+echo "============================================================================"

scripts/flexfec/run-publisher.sh

@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+#
+# Run the rust-sdks local_video publisher with FlexFEC publishing enabled.
+#
+# Connects to the local dev SFU (run-sfu.sh) and publishes camera video with a
+# FlexFEC-03 repair stream (--flex-fec). Output is teed to a log file.
+#
+# Usage:
+#   scripts/flexfec/run-publisher.sh [extra publisher args...]
+#
+# Env overrides:
+#   RUST_SDKS_DIR   path to the rust-sdks5 checkout (default: ../rust-sdks5)
+#   LIVEKIT_URL     default ws://127.0.0.1:7880
+#   LIVEKIT_API_KEY / LIVEKIT_API_SECRET   default devkey / secret
+#   ROOM            room name (default flexfec-test)
+#   PUB_IDENTITY    participant identity (default flexfec-pub)
+#   RUST_LOG        log filter (default info)
+#   LOG_DIR         where to write logs (default scripts/flexfec/logs)
+#   TEST_PATTERN    1 = SMPTE test pattern (no camera); 0 = use camera (default 1)
+#   ATTACH_META     1 = attach per-frame timestamp + frame id so the subscriber can log
+#                   end-to-end frame latency; 0 = off (default 1)
+#   MIN_PLAYOUT_DELAY / MAX_PLAYOUT_DELAY   ms; when set, recreates the room with a
+#                   subscriber playout-delay cap (passed through to the publisher)
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+RUST_SDKS_DIR="${RUST_SDKS_DIR:-$REPO_ROOT/../rust-sdks5}"
+LOG_DIR="${LOG_DIR:-$SCRIPT_DIR/logs}"
+
+if [[ ! -d "$RUST_SDKS_DIR" ]]; then
+  echo "error: rust-sdks checkout not found at '$RUST_SDKS_DIR' (set RUST_SDKS_DIR)" >&2
+  exit 1
+fi
+
+export LIVEKIT_URL="${LIVEKIT_URL:-ws://127.0.0.1:7880}"
+export LIVEKIT_API_KEY="${LIVEKIT_API_KEY:-devkey}"
+export LIVEKIT_API_SECRET="${LIVEKIT_API_SECRET:-secret}"
+export RUST_LOG="${RUST_LOG:-info}"
+ROOM="${ROOM:-flexfec-test}"
+PUB_IDENTITY="${PUB_IDENTITY:-flexfec-pub}"
+
+PUB_ARGS=(--flex-fec --room-name "$ROOM" --identity "$PUB_IDENTITY")
+if [[ "${TEST_PATTERN:-1}" == "1" ]]; then
+  PUB_ARGS+=(--test-pattern)
+fi
+if [[ "${ATTACH_META:-1}" == "1" ]]; then
+  PUB_ARGS+=(--attach-timestamp --attach-frame-id)
+fi
+if [[ -n "${MIN_PLAYOUT_DELAY:-}" ]]; then
+  PUB_ARGS+=(--min-playout-delay "$MIN_PLAYOUT_DELAY")
+fi
+if [[ -n "${MAX_PLAYOUT_DELAY:-}" ]]; then
+  PUB_ARGS+=(--max-playout-delay "$MAX_PLAYOUT_DELAY")
+fi
+
+mkdir -p "$LOG_DIR"
+echo "==> Publisher -> $LIVEKIT_URL room=$ROOM identity=$PUB_IDENTITY (FlexFEC on)"
+echo "    args: ${PUB_ARGS[*]}"
+echo "    log:  $LOG_DIR/publisher.log"
+
+cd "$RUST_SDKS_DIR"
+set -o pipefail
+cargo run -p local_video --features desktop --bin publisher -- \
+  "${PUB_ARGS[@]}" \
+  "$@" 2>&1 | tee "$LOG_DIR/publisher.log"

scripts/flexfec/run-sfu.sh

@@ -0,0 +1,62 @@
+#!/usr/bin/env bash
+#
+# Build and run the LiveKit SFU locally with FlexFEC-03 enabled.
+#
+# Uses `--dev` (devkey/secret API keys, loopback bind) together with the local
+# FlexFEC config. Media is pinned to UDP 7882 so netem.sh can shape it.
+#
+# Usage:
+#   scripts/flexfec/run-sfu.sh [extra go-run/server args...]
+#
+# Env overrides:
+#   CONFIG          path to the SFU config (default: scripts/flexfec/flexfec-local.yaml)
+#   Operator/subscriber side (SFU<->operator):
+#     DOWNLINK_LOSS      fractional SFU->subscriber loss, e.g. 0.01 for 1% (unset = none)
+#     DOWNLINK_DELAY_MS  one-way SFU->subscriber delay in ms (unset = none)
+#     UPLINK_DELAY_MS    one-way subscriber->SFU feedback (NACK/RTCP) delay in ms
+#   Robot/publisher side (robot<->SFU), e.g. a lossy 5G first hop:
+#     PUB_LOSS           fractional publisher->SFU loss, e.g. 0.03 for 3% (unset = none)
+#     PUB_DELAY_MS       one-way robot<->SFU delay in ms (inbound media + outbound NACK)
+#
+# These inject impairment in-SFU (pkg/sfu/downtrack_impair.go for the subscriber egress,
+# pkg/sfu/buffer/buffer_impair.go for the publisher ingress), replacing netem.sh for the
+# all-on-one-box setup where macOS pf/dummynet can't shape same-host UDP. Delays apply to
+# retransmissions too; set the two delays equal for a symmetric link so a NACK->RTX
+# recovery costs a full RTT -- the latency FlexFEC avoids by repairing inline.
+#
+# Teleop topology example (lossy 5G robot uplink, clean wired operator):
+#   PUB_LOSS=0.03 PUB_DELAY_MS=40 scripts/flexfec/run-sfu.sh   # + flexfec.publisher: true
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+CONFIG="${CONFIG:-$SCRIPT_DIR/flexfec-local.yaml}"
+
+cd "$REPO_ROOT"
+
+if [[ -n "${DOWNLINK_LOSS:-}" ]]; then
+  export LK_DOWNLINK_LOSS="$DOWNLINK_LOSS"
+fi
+if [[ -n "${DOWNLINK_DELAY_MS:-}" ]]; then
+  export LK_DOWNLINK_DELAY_MS="$DOWNLINK_DELAY_MS"
+fi
+if [[ -n "${UPLINK_DELAY_MS:-}" ]]; then
+  export LK_UPLINK_DELAY_MS="$UPLINK_DELAY_MS"
+fi
+if [[ -n "${PUB_LOSS:-}" ]]; then
+  export LK_PUB_LOSS="$PUB_LOSS"
+fi
+if [[ -n "${PUB_DELAY_MS:-}" ]]; then
+  export LK_PUB_DELAY_MS="$PUB_DELAY_MS"
+fi
+
+echo "==> Building & running LiveKit SFU with FlexFEC"
+echo "    repo:   $REPO_ROOT"
+echo "    config: $CONFIG"
+echo "    url:    ws://127.0.0.1:7880   (devkey / secret)"
+echo "    media:  udp/7882 on loopback"
+echo "    operator downlink: loss=${LK_DOWNLINK_LOSS:-none} delay_ms=${LK_DOWNLINK_DELAY_MS:-none} nack_delay_ms=${LK_UPLINK_DELAY_MS:-none}"
+echo "    robot uplink:      loss=${LK_PUB_LOSS:-none} delay_ms=${LK_PUB_DELAY_MS:-none}"
+echo
+
+exec go run ./cmd/server --dev --config "$CONFIG" "$@"

scripts/flexfec/run-subscriber.sh

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+#
+# Run the rust-sdks local_video subscriber against the local dev SFU and surface
+# FlexFEC stats. The subscriber logs "Video FEC ..." lines as repair packets arrive
+# and recovery payload is accepted.
+#
+# Usage:
+#   scripts/flexfec/run-subscriber.sh [extra subscriber args...]
+#
+# Env overrides:
+#   RUST_SDKS_DIR   path to the rust-sdks5 checkout (default: ../rust-sdks5)
+#   LIVEKIT_URL     default ws://127.0.0.1:7880
+#   LIVEKIT_API_KEY / LIVEKIT_API_SECRET   default devkey / secret
+#   ROOM            room name (default flexfec-test)
+#   SUB_IDENTITY    participant identity (default flexfec-sub)
+#   RUST_LOG        log filter (default info)
+#   LOG_DIR         where to write logs (default scripts/flexfec/logs)
+#   FEC_ONLY        if set to 1, only print FEC + frame-latency lines to the console
+#
+# Frame latency: when the publisher runs with --attach-timestamp/--attach-frame-id, the
+# subscriber emits "Subscriber frame latency: ..." lines (capture->receive->decode). The
+# receive_to_decode value is the jitter-buffer wait, which is where FEC-ON and RTX-only
+# diverge; capture_to_decode is the end-to-end glass-to-decode latency.
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+RUST_SDKS_DIR="${RUST_SDKS_DIR:-$REPO_ROOT/../rust-sdks5}"
+LOG_DIR="${LOG_DIR:-$SCRIPT_DIR/logs}"
+
+if [[ ! -d "$RUST_SDKS_DIR" ]]; then
+  echo "error: rust-sdks checkout not found at '$RUST_SDKS_DIR' (set RUST_SDKS_DIR)" >&2
+  exit 1
+fi
+
+export LIVEKIT_URL="${LIVEKIT_URL:-ws://127.0.0.1:7880}"
+export LIVEKIT_API_KEY="${LIVEKIT_API_KEY:-devkey}"
+export LIVEKIT_API_SECRET="${LIVEKIT_API_SECRET:-secret}"
+export RUST_LOG="${RUST_LOG:-info}"
+ROOM="${ROOM:-flexfec-test}"
+SUB_IDENTITY="${SUB_IDENTITY:-flexfec-sub}"
+
+mkdir -p "$LOG_DIR"
+echo "==> Subscriber -> $LIVEKIT_URL room=$ROOM identity=$SUB_IDENTITY"
+echo "    log: $LOG_DIR/subscriber.log   (grep 'Video FEC' for FEC, 'frame latency' for latency)"
+
+cd "$RUST_SDKS_DIR"
+set -o pipefail
+if [[ "${FEC_ONLY:-0}" == "1" ]]; then
+  cargo run -p local_video --features desktop --bin subscriber -- \
+    --room-name "$ROOM" \
+    --identity "$SUB_IDENTITY" \
+    "$@" 2>&1 | tee "$LOG_DIR/subscriber.log" | grep --line-buffered -iE 'fec|recover|frame latency' || true
+else
+  cargo run -p local_video --features desktop --bin subscriber -- \
+    --room-name "$ROOM" \
+    --identity "$SUB_IDENTITY" \
+    "$@" 2>&1 | tee "$LOG_DIR/subscriber.log"
+fi