From d269ac59dd5d9e9d7db45749fa0dc8884ea589d0 Mon Sep 17 00:00:00 2001 From: David Chen Date: Sat, 12 Sep 2026 11:44:49 -0700 Subject: [PATCH] Preserve FEC sender state and harden subscription updates Carry repair sequence state through the existing DownTrack GetState/SeedState handoff so cached transceiver reuse preserves SRTP sequence continuity. Close the old encoder before saving its sequence, and allocate its media group only when protection is first used. Apply subscriber settings from one immutable snapshot and discard superseded updates. Preserve upstream FEC on shared sendrecv codec lists when downstream generation is disabled. Lock the source RTP stats while seeding a replacement sender to avoid racing its final snapshot update. Align field and import grouping with the surrounding implementation. Add regressions for sequence wrap/rebind, SRTP sender reuse, concurrent settings, superseded updates, and independent negotiation flags. Validation: full race suites passed for pkg/sfu/flexfec, pkg/sfu/pacer, pkg/sfu/rtpstats, pkg/sfu, and pkg/rtc. Encoder benchmarks remain approximately 1.9-2.5 us per five 1200-byte media packets at low/medium/high protection; disabled encoding has zero allocations. --- pkg/rtc/subscribedtrack.go | 26 +++--- pkg/rtc/subscribedtrack_fec_test.go | 97 ++++++++++++++++++----- pkg/rtc/transport.go | 8 +- pkg/rtc/transport_fec_test.go | 21 +++++ pkg/sfu/downtrack.go | 33 +++++--- pkg/sfu/downtrack_fec.go | 33 +++++++- pkg/sfu/downtrack_fec_integration_test.go | 44 ++++++++++ pkg/sfu/downtrack_fec_test.go | 11 ++- pkg/sfu/flexfec/encoder.go | 60 ++++++++++++-- pkg/sfu/flexfec/encoder_test.go | 29 ++++++- pkg/sfu/pacer/pacer.go | 4 +- pkg/sfu/rtpstats/rtpstats_sender.go | 7 ++ 12 files changed, 311 insertions(+), 62 deletions(-) diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 54fa405f1..ea6796c13 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -137,8 +137,6 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error) // Strip packet trailer if track has packet trailer but subscriber does not have cap stripPacketTrailer := params.MediaTrack.HasPacketTrailer() && !subSupportsPacketTrailer downTrack, err := sfu.NewDownTrack(sfu.DownTrackParams{ - EnableFlexFEC: params.SubscriberConfig.FlexFEC.Enabled, - OnFECSent: prometheus.RecordFECDownstream, Codecs: codecs, IsEncrypted: isEncrypted, Source: params.MediaTrack.Source(), @@ -160,6 +158,8 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error) DisableSenderReportPassThrough: params.Subscriber.GetDisableSenderReportPassThrough(), SupportsCodecChange: params.Subscriber.SupportsCodecChange(), EnableStartAtDesiredQuality: params.EnableStartAtDesiredQuality, + EnableFlexFEC: params.SubscriberConfig.FlexFEC.Enabled, + OnFECSent: prometheus.RecordFECDownstream, Listener: s, }) if err != nil { @@ -214,6 +214,7 @@ func (t *SubscribedTrack) Bound(err error) { if t.settings != nil { if t.params.AdaptiveStream { // remove `disabled` flag to force a visibility update + t.settings = utils.CloneProto(t.settings) t.settings.Disabled = false t.logger.Debugw("enabling subscriber track settings on bind", "settings", logger.Proto(t.settings)) } @@ -352,6 +353,7 @@ func (t *SubscribedTrack) applySettings() { t.settingsVersion = t.versionGenerator.Next() settingsVersion := t.settingsVersion + settings := t.settings t.settingsLock.Unlock() dt := t.DownTrack() @@ -359,30 +361,30 @@ func (t *SubscribedTrack) applySettings() { temporal := buffer.InvalidLayerTemporal if dt.Kind() == webrtc.RTPCodecTypeVideo { mt := t.MediaTrack() - quality := t.settings.Quality + quality := settings.Quality mimeType := dt.Mime() - if t.settings.Width > 0 { - quality = mt.GetQualityForDimension(mimeType, t.settings.Width, t.settings.Height) + if settings.Width > 0 { + quality = mt.GetQualityForDimension(mimeType, settings.Width, settings.Height) } spatial = buffer.GetSpatialLayerForVideoQuality(mimeType, quality, mt.ToProto()) - if t.settings.Fps > 0 { - temporal = mt.GetTemporalLayerForSpatialFps(mimeType, spatial, t.settings.Fps) + if settings.Fps > 0 { + temporal = mt.GetTemporalLayerForSpatialFps(mimeType, spatial, settings.Fps) } } t.settingsLock.Lock() - if settingsVersion != t.settingsVersion { + if settingsVersion != t.settingsVersion || settings != t.settings { // a newer settings has superseded this one t.settingsLock.Unlock() return } - t.logger.Debugw("applying subscriber track settings", "settings", logger.Proto(t.settings)) - if t.settings.Fec != nil { - dt.SetFECProtection(*t.settings.Fec) + t.logger.Debugw("applying subscriber track settings", "settings", logger.Proto(settings)) + if settings.Fec != nil { + dt.SetFECProtection(*settings.Fec) } - if t.settings.Disabled { + if settings.Disabled { dt.Mute(true) t.settingsLock.Unlock() return diff --git a/pkg/rtc/subscribedtrack_fec_test.go b/pkg/rtc/subscribedtrack_fec_test.go index 38b3c78cc..3ad390518 100644 --- a/pkg/rtc/subscribedtrack_fec_test.go +++ b/pkg/rtc/subscribedtrack_fec_test.go @@ -15,17 +15,21 @@ package rtc import ( + "sync" "testing" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/codecs/mime" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils" + "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/sfufakes" - "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils" - "github.com/pion/webrtc/v4" - "github.com/stretchr/testify/require" ) func TestSubscriberFECSettings(t *testing.T) { @@ -80,6 +84,71 @@ func TestSubscriberFECPersistsAcrossResubscription(t *testing.T) { } func TestSubscriberFECAppliesToDownTrack(t *testing.T) { + sub, _ := newSubscriberFECTestTrack(t) + dt := sub.DownTrack() + for _, tc := range []struct { + level *livekit.FECProtection + want int64 + }{ + {nil, 100_000}, + {livekit.FECProtection_FEC_MEDIUM.Enum(), 125_000}, + {nil, 125_000}, + {livekit.FECProtection_FEC_NONE.Enum(), 100_000}, + {nil, 100_000}, + {livekit.FECProtection_FEC_LOW.Enum(), 115_000}, + {livekit.FECProtection_FEC_HIGH.Enum(), 135_000}, + {livekit.FECProtection(99).Enum(), 100_000}, + } { + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH, Fec: tc.level}, true) + dt.AllocateOptimal(false, false) + require.Equal(t, tc.want, dt.BandwidthRequested(), "client settings must reach the negotiated encoder and allocator") + } +} + +func TestSubscriberFECSettingsSupersededDuringApply(t *testing.T) { + sub, mediaTrack := newSubscriberFECTestTrack(t) + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH}, true) + var pending func() + sub.debouncer = func(f func()) { pending = f } + mediaTrack.GetQualityForDimensionCalls(func(mime.MimeType, uint32, uint32) livekit.VideoQuality { + // A settings update can arrive while layer selection runs without settingsLock. + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{ + Quality: livekit.VideoQuality_HIGH, + Fec: livekit.FECProtection_FEC_HIGH.Enum(), + }, false) + return livekit.VideoQuality_HIGH + }) + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{ + Width: 640, + Fec: livekit.FECProtection_FEC_MEDIUM.Enum(), + }, true) + dt := sub.DownTrack() + dt.AllocateOptimal(false, false) + require.EqualValues(t, 100_000, dt.BandwidthRequested(), "superseded settings must not apply part of a newer update") + require.NotNil(t, pending) + pending() + dt.AllocateOptimal(false, false) + require.EqualValues(t, 135_000, dt.BandwidthRequested()) +} + +func TestSubscriberFECConcurrentSettings(t *testing.T) { + sub, _ := newSubscriberFECTestTrack(t) + var wg sync.WaitGroup + for level := range 4 { + wg.Go(func() { + for range 100 { + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{ + Width: 640, + Fec: livekit.FECProtection(level).Enum(), + }, true) + } + }) + } + wg.Wait() +} + +func newSubscriberFECTestTrack(t *testing.T) (*SubscribedTrack, *typesfakes.FakeMediaTrack) { + t.Helper() codec := webrtc.RTPCodecParameters{ RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000}, PayloadType: 96, @@ -115,23 +184,7 @@ func TestSubscriberFECAppliesToDownTrack(t *testing.T) { logger: logger.GetLogger(), versionGenerator: utils.NewDefaultTimedVersionGenerator(), } - for _, tc := range []struct { - level *livekit.FECProtection - want int64 - }{ - {nil, 100_000}, - {livekit.FECProtection_FEC_MEDIUM.Enum(), 125_000}, - {nil, 125_000}, - {livekit.FECProtection_FEC_NONE.Enum(), 100_000}, - {nil, 100_000}, - {livekit.FECProtection_FEC_LOW.Enum(), 115_000}, - {livekit.FECProtection_FEC_HIGH.Enum(), 135_000}, - {livekit.FECProtection(99).Enum(), 100_000}, - } { - sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH, Fec: tc.level}, true) - dt.AllocateOptimal(false, false) - require.Equal(t, tc.want, dt.BandwidthRequested(), "client settings must reach the negotiated encoder and allocator") - } + return sub, mediaTrack } type subscriberFECTrackContext struct { diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 1659da944..60eac4c91 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -1022,6 +1022,12 @@ func (t *PCTransport) queueOrConfigureSender( enableAudioStereo bool, enableAudioNACK bool, ) { + keepFlexFEC := t.params.Config.Subscriber.FlexFEC.Enabled + if transceiver.Direction() == webrtc.RTPTransceiverDirectionSendrecv && t.params.DirectionConfig.FlexFEC.Enabled { + // A shared codec list must retain upstream repair support even when + // this subscription does not permit downstream FEC generation. + keepFlexFEC = true + } params := configureSenderParams{ transceiver: transceiver, enabledCodecs: enabledCodecs, @@ -1029,7 +1035,7 @@ func (t *PCTransport) queueOrConfigureSender( filterOutH264HighProfile: !t.params.IsOfferer, enableAudioStereo: enableAudioStereo, enableAudioNACK: enableAudioNACK, - keepFlexFEC: t.params.Config.Subscriber.FlexFEC.Enabled, + keepFlexFEC: keepFlexFEC, } if !t.params.IsOfferer { t.sendersPendingConfigMu.Lock() diff --git a/pkg/rtc/transport_fec_test.go b/pkg/rtc/transport_fec_test.go index a9f530657..c79ef515d 100644 --- a/pkg/rtc/transport_fec_test.go +++ b/pkg/rtc/transport_fec_test.go @@ -255,3 +255,24 @@ func TestSinglePCDownstreamFlexFEC(t *testing.T) { }) } } + +func TestSinglePCUpstreamFlexFECWithSubscription(t *testing.T) { + codecs := []*livekit.Codec{{Mime: "video/VP8"}, {Mime: "video/rtx"}} + conf := &WebRTCConfig{Publisher: DirectionConfig{FlexFEC: FlexFECDirectionConfig{Enabled: true}}} + server, err := NewPCTransport(TransportParams{ + Config: conf, DirectionConfig: conf.Publisher, IsSendSide: true, + Handler: &transportfakes.FakeHandler{}, + EnabledPublishCodecs: codecs, EnabledSubscribeCodecs: codecs, + }) + require.NoError(t, err) + defer server.Close() + tr, err := server.pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendrecv}) + require.NoError(t, err) + server.queueOrConfigureSender(tr, codecs, RTCPFeedbackConfig{}, false, true) + require.Len(t, server.sendersPendingConfig, 1) + configureSender(server.sendersPendingConfig[0], nil) + server.restrictReceiverCodecsToPublishList() + offer, err := server.pc.CreateOffer(nil) + require.NoError(t, err) + require.Contains(t, offer.SDP, "flexfec-03/90000", "subscribing on a shared m-section must retain upstream FEC when downstream FEC is disabled") +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index a90fb2397..296d2b910 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -184,6 +184,7 @@ type DownTrackState struct { DeltaStatsRTXSenderSnapshotId uint32 ForwarderState *livekit.RTPForwarderState PlayoutDelayControllerState PlayoutDelayControllerState + FECState flexfec.EncoderState } func (d DownTrackState) MarshalLogObject(e zapcore.ObjectEncoder) error { @@ -193,6 +194,7 @@ func (d DownTrackState) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddUint32("DeltaStatsRTXSenderSnapshotId", d.DeltaStatsRTXSenderSnapshotId) e.AddObject("ForwarderState", logger.Proto(d.ForwarderState)) e.AddObject("PlayoutDelayControllerState", d.PlayoutDelayControllerState) + e.AddObject("FECState", d.FECState) return nil } @@ -227,8 +229,6 @@ var _ TrackSender = (*DownTrack)(nil) type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) type DownTrackParams struct { - EnableFlexFEC bool - OnFECSent func(packets int, bytes int) Codecs []webrtc.RTPCodecParameters IsEncrypted bool Source livekit.TrackSource @@ -246,6 +246,8 @@ type DownTrackParams struct { SupportsCodecChange bool StripPacketTrailer bool EnableStartAtDesiredQuality bool + EnableFlexFEC bool + OnFECSent func(packets int, bytes int) Listener DownTrackListener } @@ -258,18 +260,20 @@ type DownTrackParams struct { // - closed // once closed, a DownTrack cannot be re-used. type DownTrack struct { + params DownTrackParams + id livekit.TrackID + kind webrtc.RTPCodecType + ssrc uint32 + ssrcRTX uint32 + payloadType atomic.Uint32 + payloadTypeRTX atomic.Uint32 + sequencer *sequencer + rtxSequenceNumber atomic.Uint64 + + fecLock sync.Mutex fecEncoder atomic.Pointer[flexfec.Encoder] fecProtectionPercent atomic.Uint32 - fecLock sync.Mutex - params DownTrackParams - id livekit.TrackID - kind webrtc.RTPCodecType - ssrc uint32 - ssrcRTX uint32 - payloadType atomic.Uint32 - payloadTypeRTX atomic.Uint32 - sequencer *sequencer - rtxSequenceNumber atomic.Uint64 + fecState flexfec.EncoderState receiverLock sync.RWMutex receiver TrackReceiver @@ -1459,6 +1463,7 @@ func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) { d.setBindStateLocked(bindStateUnbound) d.Receiver().DeleteDownTrack(d.SubscriberID()) + d.closeFEC() if d.rtcpReader != nil && isEnding { d.params.Logger.Debugw("downtrack close rtcp reader") @@ -1476,7 +1481,6 @@ func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) { d.rtpStats.Stop() d.rtpStatsRTX.Stop() - d.closeFEC() d.params.Logger.Debugw( "rtp stats", "direction", "downstream", @@ -1540,6 +1544,7 @@ func (d *DownTrack) GetState() DownTrackState { RTPStatsRTX: d.rtpStatsRTX, DeltaStatsRTXSenderSnapshotId: d.deltaStatsRTXSenderSnapshotId, ForwarderState: d.forwarder.GetState(), + FECState: d.getFECState(), } if d.playoutDelay != nil { @@ -1570,6 +1575,7 @@ func (d *DownTrack) SeedState(state DownTrackState) { d.rtxSequenceNumber.Store(d.rtpStatsRTX.ExtHighestSequenceNumber()) } d.forwarder.SeedState(state.ForwarderState) + d.seedFECState(state.FECState) } func (d *DownTrack) StopWriteAndGetState() DownTrackState { @@ -1577,6 +1583,7 @@ func (d *DownTrack) StopWriteAndGetState() DownTrackState { d.bindLock.Lock() d.writable.Store(false) d.writeStopped.Store(true) + d.closeFEC() d.bindLock.Unlock() return d.GetState() diff --git a/pkg/sfu/downtrack_fec.go b/pkg/sfu/downtrack_fec.go index 9d71ae65c..cb77aea67 100644 --- a/pkg/sfu/downtrack_fec.go +++ b/pkg/sfu/downtrack_fec.go @@ -17,24 +17,26 @@ package sfu import ( "strings" - "github.com/livekit/protocol/livekit" "github.com/pion/webrtc/v4" + "github.com/livekit/protocol/livekit" + "github.com/livekit/livekit-server/pkg/sfu/flexfec" ) // bindFEC runs under bindLock, using the complete negotiated codec list saved // before Pion narrows the TrackLocalContext to the selected media codec. func (d *DownTrack) bindFEC(t webrtc.TrackLocalContext) { - d.closeFEC() d.fecLock.Lock() defer d.fecLock.Unlock() + d.closeFECLocked() if !d.params.EnableFlexFEC || d.kind != webrtc.RTPCodecTypeVideo || t.SSRCForwardErrorCorrection() == 0 { return } for _, codec := range d.negotiatedCodecParameters { if strings.EqualFold(codec.MimeType, webrtc.MimeTypeFlexFEC03) { encoder := flexfec.NewEncoder(uint8(codec.PayloadType), uint32(t.SSRCForwardErrorCorrection()), d.params.OnFECSent) + encoder.SeedState(d.fecState) encoder.SetProtectionPercent(d.fecProtectionPercent.Load()) d.fecEncoder.Store(encoder) return @@ -45,8 +47,35 @@ func (d *DownTrack) bindFEC(t webrtc.TrackLocalContext) { func (d *DownTrack) closeFEC() { d.fecLock.Lock() defer d.fecLock.Unlock() + d.closeFECLocked() +} + +func (d *DownTrack) closeFECLocked() { if encoder := d.fecEncoder.Swap(nil); encoder != nil { encoder.Close() + // Close first so queued media cannot advance the sequence after it is saved. + d.fecState = encoder.GetState() + } +} + +func (d *DownTrack) getFECState() flexfec.EncoderState { + d.fecLock.Lock() + defer d.fecLock.Unlock() + if encoder := d.fecEncoder.Load(); encoder != nil { + return encoder.GetState() + } + return d.fecState +} + +func (d *DownTrack) seedFECState(state flexfec.EncoderState) { + if state.SSRC == 0 { + return + } + d.fecLock.Lock() + defer d.fecLock.Unlock() + d.fecState = state + if encoder := d.fecEncoder.Load(); encoder != nil { + encoder.SeedState(state) } } diff --git a/pkg/sfu/downtrack_fec_integration_test.go b/pkg/sfu/downtrack_fec_integration_test.go index e8b550a6e..95abba6e1 100644 --- a/pkg/sfu/downtrack_fec_integration_test.go +++ b/pkg/sfu/downtrack_fec_integration_test.go @@ -247,6 +247,50 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { } } } + + // Resume a subscription on its cached transceiver, including the + // repair sequence. The receiver's existing SRTP replay window must + // accept every packet without another negotiation. + dt.CloseWithFlush(false, false) + state := dt.GetState() + require.Equal(t, sequence+1, state.FECState.NextSequenceNumber) + next, err := sfu.NewDownTrack(sfu.DownTrackParams{ + Codecs: []webrtc.RTPCodecParameters{vp8CodecParams}, + Receiver: newFakeTrackReceiver(vp8CodecParams), + BufferFactory: factory, + Pacer: p, + Logger: logger.GetLogger(), + Listener: &sfufakes.FakeDownTrackListener{}, + StreamID: "fec-stream", + SubID: "fec-sub", + MaxTrack: 500, + EnableFlexFEC: true, + }) + require.NoError(t, err) + next.OnBinding(func(err error) { + if err == nil { + next.SeedState(state) + } + }) + next.SetTransceiver(tr) + next.SetFECProtection(livekit.FECProtection_FEC_MEDIUM) + require.NoError(t, tr.Sender().ReplaceTrack(next)) + dt = next + dt.SetConnected() + require.Eventually(t, dt.IsWritableForTest, 5*time.Second, 10*time.Millisecond) + dt.ForceForwardLayerForTest(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + before := capture.count() + sendMedia(start, mediaCount) + require.Eventually(t, func() bool { return capture.count() >= before+25 }, 5*time.Second, 10*time.Millisecond) + repairs := 0 + for _, packet := range capture.all()[before:] { + if packet.SSRC == fecSSRC { + require.Equal(t, sequence+1, packet.SequenceNumber) + sequence = packet.SequenceNumber + repairs++ + } + } + require.Equal(t, 5, repairs) } }) } diff --git a/pkg/sfu/downtrack_fec_test.go b/pkg/sfu/downtrack_fec_test.go index 651c9fb01..1fe62ad0b 100644 --- a/pkg/sfu/downtrack_fec_test.go +++ b/pkg/sfu/downtrack_fec_test.go @@ -17,11 +17,13 @@ package sfu import ( "testing" - "github.com/livekit/livekit-server/pkg/sfu/flexfec" - "github.com/livekit/protocol/livekit" "github.com/pion/rtp" "github.com/pion/webrtc/v4" "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" + + "github.com/livekit/livekit-server/pkg/sfu/flexfec" ) type fecTrackContext struct { @@ -54,6 +56,7 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) { old := d.fecEncoder.Load() require.NotNil(t, old) p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96} + var lastSequenceNumber uint16 for i := range flexfec.MediaPacketsPerGroup { p.SequenceNumber++ repair := old.Encode(&p, []byte{1, 2, 3}) @@ -61,6 +64,7 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) { require.Len(t, repair, 1) require.EqualValues(t, 456, repair[0].SSRC) require.EqualValues(t, 118, repair[0].PayloadType, "use negotiated PT") + lastSequenceNumber = repair[0].SequenceNumber } } d.bindFEC(c) @@ -68,6 +72,9 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) { for range flexfec.MediaPacketsPerGroup { p.SequenceNumber++ require.Empty(t, old.Encode(&p, []byte{1, 2, 3}), "old queued writes cannot generate repair") + for _, repair := range d.fecEncoder.Load().Encode(&p, []byte{1, 2, 3}) { + require.Equal(t, lastSequenceNumber+1, repair.SequenceNumber, "reusing an SSRC must preserve repair sequencing") + } } d.closeFEC() require.Nil(t, d.fecEncoder.Load()) diff --git a/pkg/sfu/flexfec/encoder.go b/pkg/sfu/flexfec/encoder.go index 60de5c631..293dbd5a2 100644 --- a/pkg/sfu/flexfec/encoder.go +++ b/pkg/sfu/flexfec/encoder.go @@ -21,6 +21,7 @@ import ( pionflexfec "github.com/pion/interceptor/pkg/flexfec" "github.com/pion/rtp" + "go.uber.org/zap/zapcore" "github.com/livekit/protocol/utils/mono" ) @@ -37,6 +38,24 @@ const ( maxEncoderMediaPacketSize = maxMediaPacketSize - pionflexfec.BaseFec03HeaderSize - 16 ) +// EncoderState preserves the repair sequence when a sender SSRC is reused. +// Partial media groups are never carried over to a new binding. +type EncoderState struct { + SSRC uint32 + NextSequenceNumber uint16 +} + +func (s EncoderState) MarshalLogObject(e zapcore.ObjectEncoder) error { + e.AddUint32("SSRC", s.SSRC) + e.AddUint16("NextSequenceNumber", s.NextSequenceNumber) + return nil +} + +type encoderMediaGroup struct { + packets [MediaPacketsPerGroup]rtp.Packet + storage [MediaPacketsPerGroup][maxEncoderMediaPacketSize]byte +} + // Encoder batches final, outgoing video packets for Pion's FlexFEC-03 encoder. // It owns reusable wire storage: the caller may recycle headers, extensions and // payloads as soon as Encode returns. It does not delay media or start workers. @@ -44,11 +63,11 @@ const ( type Encoder struct { mu sync.Mutex encoder *pionflexfec.FlexEncoder03 - packets [MediaPacketsPerGroup]rtp.Packet - storage [MediaPacketsPerGroup][maxEncoderMediaPacketSize]byte + media *encoderMediaGroup count int startedAt int64 sequenceNumber uint16 + ssrc uint32 closed bool onSent func(packets int, bytes int) protectionPercent uint32 @@ -59,11 +78,31 @@ func NewEncoder(payloadType uint8, ssrc uint32, onSent func(packets int, bytes i return &Encoder{ encoder: pionflexfec.NewFlexEncoder03(payloadType, ssrc), sequenceNumber: uint16(rand.Uint32()), + ssrc: ssrc, onSent: onSent, protectionPercent: DefaultProtectionPercent, } } +func (e *Encoder) GetState() EncoderState { + e.mu.Lock() + defer e.mu.Unlock() + return EncoderState{SSRC: e.ssrc, NextSequenceNumber: e.sequenceNumber} +} + +// SeedState must be called before forwarding starts. Only reuse sequencing for +// the same repair SSRC; a new SSRC keeps its randomized initial sequence number. +func (e *Encoder) SeedState(state EncoderState) { + e.mu.Lock() + defer e.mu.Unlock() + if e.closed || state.SSRC == 0 || state.SSRC != e.ssrc { + return + } + e.sequenceNumber = state.NextSequenceNumber + e.count = 0 + e.repairCredit = 0 +} + // Encode returns independently owned repair packets for complete groups only. // Gaps (including skipped padding), codec switches and stale groups start a new // group, since Pion requires consecutive sequence numbers. RTP wrap is valid. @@ -81,8 +120,13 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { e.count = 0 return nil } + if e.media == nil { + // Like the upstream recovery buffer, allocate packet storage only once + // it is needed. Negotiated tracks with the default "none" stay small. + e.media = &encoderMediaGroup{} + } if e.count != 0 { - previous := &e.packets[e.count-1] + previous := &e.media.packets[e.count-1] if header.SequenceNumber != previous.SequenceNumber+1 || header.SSRC != previous.SSRC || header.PayloadType != previous.PayloadType || now-e.startedAt > int64(maxEncoderGroupAge) { @@ -93,14 +137,14 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { e.startedAt = now } - raw := e.storage[e.count][:size] + raw := e.media.storage[e.count][:size] n, err := header.MarshalTo(raw) if err != nil { e.count = 0 return nil } copy(raw[n:], payload) - if err = e.packets[e.count].Unmarshal(raw); err != nil { + if err = e.media.packets[e.count].Unmarshal(raw); err != nil { e.count = 0 return nil } @@ -119,9 +163,9 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { if numRepair == 0 { return nil } - repair := e.encoder.EncodeFec(e.packets[:], numRepair) + repair := e.encoder.EncodeFec(e.media.packets[:], numRepair) for i := range repair { - // Pion's encoder uses a constant timestamp and initial sequence number. + // Pion's encoder uses a constant timestamp and a fixed initial sequence number. // Use this stream's media clock and a randomized, continuous repair sequence. repair[i].Timestamp = header.Timestamp repair[i].SequenceNumber = e.sequenceNumber @@ -157,5 +201,7 @@ func (e *Encoder) Close() { e.mu.Lock() e.closed = true e.count = 0 + e.media = nil + e.encoder = nil e.mu.Unlock() } diff --git a/pkg/sfu/flexfec/encoder_test.go b/pkg/sfu/flexfec/encoder_test.go index 20188cefb..f4e1d4688 100644 --- a/pkg/sfu/flexfec/encoder_test.go +++ b/pkg/sfu/flexfec/encoder_test.go @@ -19,9 +19,10 @@ import ( "sync" "testing" - "github.com/livekit/protocol/logger" "github.com/pion/rtp" "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/logger" ) func TestEncoderDefaultsToNoProtection(t *testing.T) { @@ -32,6 +33,32 @@ func TestEncoderDefaultsToNoProtection(t *testing.T) { require.Empty(t, encoder.Encode(&p, []byte{1, 2, 3})) } require.Zero(t, encoder.count, "disabled protection must not retain media") + require.Nil(t, encoder.media, "negotiation alone must not allocate packet storage") +} + +func TestEncoderState(t *testing.T) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(20) + e.SeedState(EncoderState{SSRC: testFECSSRC, NextSequenceNumber: 65535}) + for _, sn := range []uint16{65535, 0} { + media := makeMediaPackets(t, 100, MediaPacketsPerGroup) + for i := range media { + repair := e.Encode(&media[i].Header, media[i].Payload) + if i == MediaPacketsPerGroup-1 { + require.Len(t, repair, 1) + require.Equal(t, sn, repair[0].SequenceNumber) + } + } + e.Close() + state := e.GetState() + require.Equal(t, sn+1, state.NextSequenceNumber) + e = NewEncoder(testFECPT, testFECSSRC, nil) + e.SeedState(state) + e.SetProtectionPercent(20) + } + state := e.GetState() + e.SeedState(EncoderState{SSRC: testFECSSRC + 1, NextSequenceNumber: state.NextSequenceNumber + 100}) + require.Equal(t, state, e.GetState(), "do not seed a different repair SSRC") } func TestEncoderRecoveryWithReusedMemory(t *testing.T) { diff --git a/pkg/sfu/pacer/pacer.go b/pkg/sfu/pacer/pacer.go index bf02ce988..88a2edac5 100644 --- a/pkg/sfu/pacer/pacer.go +++ b/pkg/sfu/pacer/pacer.go @@ -43,8 +43,6 @@ const ( ) type Packet struct { - // FEC is set only on original video media, never padding, RTX or probes. - FEC *flexfec.Encoder Header *rtp.Header HeaderPool *sync.Pool HeaderSize int @@ -57,6 +55,8 @@ type Packet struct { WriteStream webrtc.TrackLocalWriter Pool *sync.Pool PoolEntity *[]byte + // FEC is set only on original video media, never padding, RTX or probes. + FEC *flexfec.Encoder } type Pacer interface { diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index eafa7bc7b..09f37c6e7 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -394,8 +394,15 @@ func NewRTPStatsSender(params RTPStatsParams, cacheSize int) *RTPStatsSender { } func (r *RTPStatsSender) Seed(from *RTPStatsSender) { + if from == nil || from == r { + return + } r.lock.Lock() defer r.lock.Unlock() + // The previous sender's final stats callback can still update snapshots + // after Stop. Serialize the state copy with those updates. + from.lock.RLock() + defer from.lock.RUnlock() if !r.seed(from.rtpStatsBase) { return