diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index c15d2e1d4..a28ea20f5 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -530,6 +530,8 @@ func newPeerConnection( params.Logger.Debugw("rtx pair found from extension", "repair", repair, "base", base, "rsid", rsid) params.Config.BufferFactory.SetRTXPair(repair, base, rsid) }, + params.Config.BufferFactory, + params.SimTracks, params.Logger, ) // put rtx interceptor behind unhandle simulcast interceptor so it can get the correct mid & rid diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index c7383c7b4..5468b3b94 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -72,6 +72,24 @@ type Buffer struct { primaryBufferForRTX *Buffer rtxPktBuf []byte + + streamInfoProbe *StreamInfoProbe + warnedPendingOverflow bool +} + +// StreamInfoProbe identifies a stream from the mid/rid/rsid header extensions of its +// packets. It runs on the write path, i. e. as SRTP pushes into this buffer, because +// nothing reads remote streams through pion's interceptor chain. +type StreamInfoProbe struct { + MidExtID uint8 + RidExtID uint8 + RsidExtID uint8 + + // Tries bounds how many packets are inspected before giving up. + Tries int + + // OnFound is called at most once, in a goroutine, as it can re-enter this buffer. + OnFound func(ssrc uint32, mid, rid, rsid string) } func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer { @@ -166,6 +184,10 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { return } + if b.streamInfoProbe != nil { + b.probeStreamInfoLocked(&rtpPacket) + } + // handle RTX packet if pb := b.primaryBufferForRTX; pb != nil { b.Unlock() @@ -191,6 +213,17 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { overflow := len(b.pPackets) - max(b.BufferBase.MaxVideoPkts(), b.BufferBase.MaxAudioPkts()) if overflow > 0 { startIdx = overflow + + // a stream that keeps arriving but never binds drops every packet from here + // on; for an RTX stream it means the pairing was never established + if !b.warnedPendingOverflow { + b.warnedPendingOverflow = true + b.logger.Warnw( + "unbound buffer overflowing, dropping packets", nil, + "ssrc", b.BufferBase.SSRC(), + "pending", len(b.pPackets), + ) + } } b.pPackets = append(b.pPackets[startIdx:], pendingPacket{ packet: packet, @@ -213,6 +246,59 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { return } +// SetStreamInfoProbe installs probe and runs it over packets already queued. +func (b *Buffer) SetStreamInfoProbe(probe *StreamInfoProbe) { + b.Lock() + defer b.Unlock() + + b.streamInfoProbe = probe + + for _, pp := range b.pPackets { + if b.streamInfoProbe == nil { + return + } + + var rtpPacket rtp.Packet + if err := rtpPacket.Unmarshal(pp.packet); err != nil { + continue + } + b.probeStreamInfoLocked(&rtpPacket) + } +} + +// probeStreamInfoLocked inspects one packet, clearing the probe once the stream is +// identified or the try budget runs out. +func (b *Buffer) probeStreamInfoLocked(rtpPacket *rtp.Packet) { + probe := b.streamInfoProbe + + var mid, rid, rsid string + if ext := rtpPacket.GetExtension(probe.MidExtID); ext != nil { + mid = string(ext) + } + if ext := rtpPacket.GetExtension(probe.RidExtID); ext != nil { + rid = string(ext) + } + if ext := rtpPacket.GetExtension(probe.RsidExtID); ext != nil { + rsid = string(ext) + } + + if mid != "" && (rid != "" || rsid != "") { + b.streamInfoProbe = nil + b.logger.Debugw("stream found", "ssrc", rtpPacket.SSRC, "mid", mid, "rid", rid, "rsid", rsid) + go probe.OnFound(rtpPacket.SSRC, mid, rid, rsid) + return + } + + // ignore padding only packets for probe count + if rtpPacket.Padding && len(rtpPacket.Payload) == 0 { + return + } + + if probe.Tries--; probe.Tries <= 0 { + b.streamInfoProbe = nil + } +} + func (b *Buffer) SetPrimaryBufferForRTX(primaryBuffer *Buffer) { b.Lock() b.primaryBufferForRTX = primaryBuffer diff --git a/pkg/sfu/buffer/factory.go b/pkg/sfu/buffer/factory.go index 4a73ab057..8fb6d46dd 100644 --- a/pkg/sfu/buffer/factory.go +++ b/pkg/sfu/buffer/factory.go @@ -118,6 +118,21 @@ func (f *Factory) GetRTCPReader(ssrc uint32) *RTCPReader { return f.rtcpReaders[ssrc] } +// SetStreamInfoProbe installs probe on the buffer of ssrc, reporting whether that +// buffer exists. False means the stream can never be identified. +func (f *Factory) SetStreamInfoProbe(ssrc uint32, probe *StreamInfoProbe) bool { + f.RLock() + buffer := f.rtpBuffers[ssrc] + f.RUnlock() + + if buffer == nil { + return false + } + + buffer.SetStreamInfoProbe(probe) + return true +} + func (f *Factory) SetRTXPair(repair, base uint32, rsid string) { f.Lock() repairBuffer, baseBuffer := f.rtpBuffers[repair], f.rtpBuffers[base] diff --git a/pkg/sfu/downtrack_downstream_integration_test.go b/pkg/sfu/downtrack_downstream_integration_test.go index c711efdbf..1f946a531 100644 --- a/pkg/sfu/downtrack_downstream_integration_test.go +++ b/pkg/sfu/downtrack_downstream_integration_test.go @@ -45,12 +45,8 @@ import ( "testing" "time" - "github.com/pion/interceptor" - "github.com/pion/logging" "github.com/pion/rtcp" "github.com/pion/rtp" - "github.com/pion/sdp/v3" - "github.com/pion/transport/v4/vnet" "github.com/pion/webrtc/v4" "github.com/stretchr/testify/require" @@ -70,165 +66,9 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/sfufakes" "github.com/livekit/livekit-server/pkg/sfu/streamallocator" "github.com/livekit/livekit-server/pkg/sfu/testutils" + "github.com/livekit/livekit-server/pkg/testutils/vnettest" ) -// ----------------------------------------------------------------------------- -// vnet harness -// ----------------------------------------------------------------------------- - -type vnetHarness struct { - wan *vnet.Router - offerNet *vnet.Net - answerNet *vnet.Net -} - -func buildVNet(t *testing.T) *vnetHarness { - t.Helper() - - wan, err := vnet.NewRouter(&vnet.RouterConfig{ - CIDR: "1.2.3.0/24", - LoggerFactory: logging.NewDefaultLoggerFactory(), - }) - require.NoError(t, err) - - offerNet, err := vnet.NewNet(&vnet.NetConfig{StaticIPs: []string{"1.2.3.4"}}) - require.NoError(t, err) - require.NoError(t, wan.AddNet(offerNet)) - - answerNet, err := vnet.NewNet(&vnet.NetConfig{StaticIPs: []string{"1.2.3.5"}}) - require.NoError(t, err) - require.NoError(t, wan.AddNet(answerNet)) - - require.NoError(t, wan.Start()) - t.Cleanup(func() { _ = wan.Stop() }) - - return &vnetHarness{wan: wan, offerNet: offerNet, answerNet: answerNet} -} - -// mediaEngineConfig describes what codecs / header extensions to register on a PC. -type mediaEngineConfig struct { - video bool // register VP8 (+ RTX); otherwise register opus - headerExtensions bool // register abs-send-time + transport-cc header extensions -} - -func newMediaPC(t *testing.T, net *vnet.Net, factory *buffer.Factory, cfg mediaEngineConfig) *webrtc.PeerConnection { - t.Helper() - - me := &webrtc.MediaEngine{} - if cfg.video { - require.NoError(t, me.RegisterCodec(webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeVP8, ClockRate: 90000, RTCPFeedback: videoRTCPFeedback(), - }, - PayloadType: 96, - }, webrtc.RTPCodecTypeVideo)) - require.NoError(t, me.RegisterCodec(webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeRTX, ClockRate: 90000, SDPFmtpLine: "apt=96", - }, - PayloadType: 97, - }, webrtc.RTPCodecTypeVideo)) - } else { - require.NoError(t, me.RegisterCodec(webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{ - MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2, - }, - PayloadType: 111, - }, webrtc.RTPCodecTypeAudio)) - } - - if cfg.headerExtensions { - kind := webrtc.RTPCodecTypeAudio - if cfg.video { - kind = webrtc.RTPCodecTypeVideo - } - require.NoError(t, me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: sdp.ABSSendTimeURI}, kind)) - require.NoError(t, me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: sdp.TransportCCURI}, kind)) - } - - // no pion default interceptors: the DownTrack/pacer fill abs-send-time and - // transport-cc themselves, and the tests drive RTCP feedback synthetically, so - // pion's own feedback generators would only add nondeterminism. - ir := &interceptor.Registry{} - - se := webrtc.SettingEngine{} - se.SetNet(net) - se.SetICETimeouts(time.Second, time.Second, 200*time.Millisecond) - se.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeUDP4}) - if factory != nil { - se.BufferFactory = factory.GetOrNew - } - - api := webrtc.NewAPI( - webrtc.WithMediaEngine(me), - webrtc.WithInterceptorRegistry(ir), - webrtc.WithSettingEngine(se), - ) - pc, err := api.NewPeerConnection(webrtc.Configuration{}) - require.NoError(t, err) - t.Cleanup(func() { _ = pc.Close() }) - - return pc -} - -func videoRTCPFeedback() []webrtc.RTCPFeedback { - return []webrtc.RTCPFeedback{ - {Type: "nack"}, - {Type: "nack", Parameter: "pli"}, - {Type: webrtc.TypeRTCPFBTransportCC}, - {Type: webrtc.TypeRTCPFBGoogREMB}, - } -} - -// signalPair performs a full offer/answer exchange between two PCs (adapted from -// pion's own test helper) and waits for both to reach the connected state. -func signalPair(t *testing.T, offerer, answerer *webrtc.PeerConnection) { - t.Helper() - - connected := untilConnected(offerer, answerer) - - offer, err := offerer.CreateOffer(nil) - require.NoError(t, err) - gatherOffer := webrtc.GatheringCompletePromise(offerer) - require.NoError(t, offerer.SetLocalDescription(offer)) - <-gatherOffer - - require.NoError(t, answerer.SetRemoteDescription(*offerer.LocalDescription())) - - answer, err := answerer.CreateAnswer(nil) - require.NoError(t, err) - gatherAnswer := webrtc.GatheringCompletePromise(answerer) - require.NoError(t, answerer.SetLocalDescription(answer)) - <-gatherAnswer - - require.NoError(t, offerer.SetRemoteDescription(*answerer.LocalDescription())) - - select { - case <-connected: - case <-time.After(10 * time.Second): - t.Fatal("timed out waiting for peer connections to connect") - } -} - -func untilConnected(pcs ...*webrtc.PeerConnection) <-chan struct{} { - var wg sync.WaitGroup - wg.Add(len(pcs)) - for _, pc := range pcs { - var once sync.Once - pc.OnConnectionStateChange(func(s webrtc.PeerConnectionState) { - if s == webrtc.PeerConnectionStateConnected { - once.Do(wg.Done) - } - }) - } - done := make(chan struct{}) - go func() { - wg.Wait() - close(done) - }() - return done -} - // ----------------------------------------------------------------------------- // packet capture on the far side // ----------------------------------------------------------------------------- @@ -320,7 +160,7 @@ var ( PayloadType: 111, } vp8CodecParams = webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000, RTCPFeedback: videoRTCPFeedback()}, + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000, RTCPFeedback: vnettest.VideoRTCPFeedback()}, PayloadType: 96, } ) @@ -336,11 +176,11 @@ type downTrackHarness struct { // newBoundDownTrack builds a real DownTrack, attaches it to a real sender PC, // negotiates with a real subscriber PC over vnet, and makes it writable. -func newBoundDownTrack(t *testing.T, h *vnetHarness, factory *buffer.Factory, codecParams webrtc.RTPCodecParameters, p pacer.Pacer, meCfg mediaEngineConfig) *downTrackHarness { +func newBoundDownTrack(t *testing.T, h *vnettest.Hosts, factory *buffer.Factory, codecParams webrtc.RTPCodecParameters, p pacer.Pacer, meCfg vnettest.MediaEngineConfig) *downTrackHarness { t.Helper() - sender := newMediaPC(t, h.offerNet, factory, meCfg) - sub := newMediaPC(t, h.answerNet, factory, meCfg) + sender := vnettest.NewPeerConnection(t, vnettest.PCConfig{Net: h.OfferNet, MediaEngine: meCfg, BufferFactory: factory.GetOrNew}) + sub := vnettest.NewPeerConnection(t, vnettest.PCConfig{Net: h.AnswerNet, MediaEngine: meCfg, BufferFactory: factory.GetOrNew}) capture := captureTrack(sub) rcv := newFakeTrackReceiver(codecParams) @@ -361,7 +201,7 @@ func newBoundDownTrack(t *testing.T, h *vnetHarness, factory *buffer.Factory, co require.NoError(t, err) dt.SetTransceiver(tr) - signalPair(t, sender, sub) + vnettest.SignalPair(t, sender, sub) dt.SetConnected() require.Eventually(t, func() bool { @@ -393,10 +233,10 @@ func distinctivePayload(seed byte, n int) []byte { // ----------------------------------------------------------------------------- func TestPionVNetForwardingSpike(t *testing.T) { - h := buildVNet(t) + h := vnettest.NewHosts(t) - sender := newMediaPC(t, h.offerNet, nil, mediaEngineConfig{video: true}) - receiver := newMediaPC(t, h.answerNet, nil, mediaEngineConfig{video: true}) + sender := vnettest.NewPeerConnection(t, vnettest.PCConfig{Net: h.OfferNet, MediaEngine: vnettest.MediaEngineConfig{Video: true}}) + receiver := vnettest.NewPeerConnection(t, vnettest.PCConfig{Net: h.AnswerNet, MediaEngine: vnettest.MediaEngineConfig{Video: true}}) track, err := webrtc.NewTrackLocalStaticRTP( webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000}, @@ -408,7 +248,7 @@ func TestPionVNetForwardingSpike(t *testing.T) { cap := captureTrack(receiver) - signalPair(t, sender, receiver) + vnettest.SignalPair(t, sender, receiver) // pump a few RTP packets from the sender track go func() { @@ -445,12 +285,12 @@ func TestPionVNetForwardingSpike(t *testing.T) { // contiguous sequence numbers, timestamps, payload bytes, and that the padding bit // is cleared regardless of the source packet's padding bit. func TestDownTrackForwardsMedia(t *testing.T) { - h := buildVNet(t) + h := vnettest.NewHosts(t) factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() p := pacer.NewPassThrough(logger.GetLogger(), newNullBWE()) t.Cleanup(p.Stop) - dh := newBoundDownTrack(t, h, factory, opusCodecParams, p, mediaEngineConfig{video: false}) + dh := newBoundDownTrack(t, h, factory, opusCodecParams, p, vnettest.MediaEngineConfig{}) const numPackets = 20 sn := uint16(23333) @@ -517,12 +357,12 @@ func TestDownTrackForwardsMedia(t *testing.T) { // type, sequence number, timestamp, payload bytes, and that the padding bit is // cleared regardless of the source packet's padding bit. func TestDownTrackRetransmitsPacketsAsIs(t *testing.T) { - h := buildVNet(t) + h := vnettest.NewHosts(t) factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() p := pacer.NewPassThrough(logger.GetLogger(), newNullBWE()) t.Cleanup(p.Stop) - dh := newBoundDownTrack(t, h, factory, opusCodecParams, p, mediaEngineConfig{video: false}) + dh := newBoundDownTrack(t, h, factory, opusCodecParams, p, vnettest.MediaEngineConfig{}) const numPackets = 10 targetSN := uint16(40000) @@ -597,13 +437,13 @@ func TestDownTrackRetransmitsPacketsAsIs(t *testing.T) { // fields are not observable on the far side. The packets are still handed to the real // pacer (and written over pion). func TestDownTrackRetransmitsPacketsViaRTX(t *testing.T) { - h := buildVNet(t) + h := vnettest.NewHosts(t) factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() cp := &capturingPacer{inner: pacer.NewPassThrough(logger.GetLogger(), newNullBWE())} t.Cleanup(cp.Stop) // VP8 registers an RTX codec, so the bound DownTrack has an RTX SSRC / payload type. - dh := newBoundDownTrack(t, h, factory, vp8CodecParams, cp, mediaEngineConfig{video: true}) + dh := newBoundDownTrack(t, h, factory, vp8CodecParams, cp, vnettest.MediaEngineConfig{Video: true}) require.NotZero(t, dh.dt.SSRCRTX(), "RTX ssrc should be negotiated") require.NotZero(t, dh.dt.PayloadTypeRTXForTest(), "RTX payload type should be negotiated") @@ -680,12 +520,12 @@ func TestDownTrackRetransmitsPacketsViaRTX(t *testing.T) { // bytes removed, i. e. the retransmitted media payload is byte identical to the // original transmission. func TestDownTrackReplaysPacketTrailerStripOnRetransmit(t *testing.T) { - h := buildVNet(t) + h := vnettest.NewHosts(t) factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() cp := &capturingPacer{inner: pacer.NewPassThrough(logger.GetLogger(), newNullBWE())} t.Cleanup(cp.Stop) - dh := newBoundDownTrack(t, h, factory, vp8CodecParams, cp, mediaEngineConfig{video: true}) + dh := newBoundDownTrack(t, h, factory, vp8CodecParams, cp, vnettest.MediaEngineConfig{Video: true}) require.NotZero(t, dh.dt.SSRCRTX(), "RTX ssrc should be negotiated") video := distinctivePayload(11, 40) @@ -749,12 +589,12 @@ func lktsTrailer() []byte { // and reports both the padding bit and the padding size, so the received payload // length equals the declared padding size (RTPPaddingMaxPayloadSize). func TestDownTrackSendsPaddingOnlyPackets(t *testing.T) { - h := buildVNet(t) + h := vnettest.NewHosts(t) factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() p := pacer.NewPassThrough(logger.GetLogger(), newNullBWE()) t.Cleanup(p.Stop) - dh := newBoundDownTrack(t, h, factory, vp8CodecParams, p, mediaEngineConfig{video: true}) + dh := newBoundDownTrack(t, h, factory, vp8CodecParams, p, vnettest.MediaEngineConfig{Video: true}) // force a valid target/current layer so the video forwarder will forward (test seam // also used by forwarder_test.go's disable()). @@ -971,7 +811,7 @@ func TestDownTrackSendsProbePackets(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - h := buildVNet(t) + h := vnettest.NewHosts(t) factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() b := tc.makeBWE() @@ -981,9 +821,9 @@ func TestDownTrackSendsProbePackets(t *testing.T) { // header extensions (abs-send-time for remote BWE, transport-cc for send-side) // are needed for the DownTrack to pick up an ext id, which WriteProbePackets // requires. - dh := newBoundDownTrack(t, h, factory, vp8CodecParams, cp, mediaEngineConfig{ - video: true, - headerExtensions: true, + dh := newBoundDownTrack(t, h, factory, vp8CodecParams, cp, vnettest.MediaEngineConfig{ + Video: true, + HeaderExtensions: true, }) require.NotZero(t, dh.dt.SSRCRTX(), "RTX ssrc should be negotiated") diff --git a/pkg/sfu/interceptor/rtx.go b/pkg/sfu/interceptor/rtx.go index 1451ffa29..e6f4f2c57 100644 --- a/pkg/sfu/interceptor/rtx.go +++ b/pkg/sfu/interceptor/rtx.go @@ -21,6 +21,7 @@ import ( "github.com/pion/sdp/v3" "github.com/pion/webrtc/v4" + "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/utils" "github.com/livekit/protocol/logger" ) @@ -31,6 +32,12 @@ const ( rtxProbeCount = 10 ) +// StreamInfoProber installs a bounded probe that identifies a remote stream from the +// mid/rid/rsid header extensions of its packets. Implemented by buffer.Factory. +type StreamInfoProber interface { + SetStreamInfoProbe(ssrc uint32, probe *buffer.StreamInfoProbe) bool +} + type streamInfo struct { mid string rid string @@ -40,22 +47,51 @@ type streamInfo struct { type RTXInfoExtractorFactory struct { onStreamFound func(*interceptor.StreamInfo) onRTXPairFound func(repair, base uint32, rsid string) + prober StreamInfoProber lock sync.Mutex streams map[uint32]streamInfo + paired map[uint32]struct{} logger logger.Logger } func NewRTXInfoExtractorFactory( onStreamFound func(*interceptor.StreamInfo), onRTXPairFound func(repair, base uint32, rsid string), + prober StreamInfoProber, + simTracks map[uint32]SimulcastTrackInfo, logger logger.Logger, ) *RTXInfoExtractorFactory { - return &RTXInfoExtractorFactory{ + f := &RTXInfoExtractorFactory{ onStreamFound: onStreamFound, onRTXPairFound: onRTXPairFound, + prober: prober, streams: make(map[uint32]streamInfo), + paired: make(map[uint32]struct{}), logger: logger, } + f.seedSimulcastTracks(simTracks) + return f +} + +// seedSimulcastTracks pairs migrated streams from the migration info. A migrated client +// is mid-stream and stops sending rid/rsid, so the extensions never appear on the wire +// and the pairing has to come from what is already known about the tracks. +func (f *RTXInfoExtractorFactory) seedSimulcastTracks(simTracks map[uint32]SimulcastTrackInfo) { + for ssrc, info := range simTracks { + if info.Mid == "" || info.StreamID == "" { + continue + } + + if info.IsRepairStream { + f.SetStreamInfo(ssrc, info.Mid, "", info.StreamID) + continue + } + + f.SetStreamInfo(ssrc, info.Mid, info.StreamID, "") + if info.RepairSSRC != 0 { + f.SetStreamInfo(info.RepairSSRC, info.Mid, "", info.StreamID) + } + } } func (f *RTXInfoExtractorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) { @@ -75,6 +111,12 @@ func (f *RTXInfoExtractorFactory) SetStreamInfo(ssrc uint32, mid, rid, rsid stri return } + // the same stream can be reported by both the packet probe and the migration info + if _, ok := f.paired[ssrc]; ok { + f.lock.Unlock() + return + } + if rsid != "" { // repair stream found, find base stream for base, info := range f.streams { @@ -106,13 +148,15 @@ func (f *RTXInfoExtractorFactory) SetStreamInfo(ssrc uint32, mid, rid, rsid stri rid: rid, rsid: rsid, } + f.lock.Unlock() + return } + f.paired[repairSsrc] = struct{}{} + f.paired[baseSsrc] = struct{}{} f.lock.Unlock() - if repairSsrc != 0 && baseSsrc != 0 { - f.onRTXPairFound(repairSsrc, baseSsrc, repairSid) - } + f.onRTXPairFound(repairSsrc, baseSsrc, repairSid) } // ------------------------------------------ @@ -134,65 +178,25 @@ func (u *RTXInfoExtractor) BindRemoteStream(info *interceptor.StreamInfo, reader return reader } - return &rtxInfoReader{ - tryTimes: rtxProbeCount, - reader: reader, - midExtID: uint8(midExtensionID), - ridExtID: uint8(streamIDExtensionID), - rsidExtID: uint8(repairStreamIDExtensionID), - factory: u.factory, - logger: u.logger, + // Probe on the buffer write path rather than by wrapping this reader. Remote streams + // are consumed through SettingEngine.BufferFactory, so nothing here reads the + // interceptor chain. pion used to drive the repair stream reader, but since + // pion/webrtc#3470 it only does so when the application reads the TrackRemote or + // when no BufferFactory is set, neither of which holds. + ok := u.factory.prober.SetStreamInfoProbe(info.SSRC, &buffer.StreamInfoProbe{ + MidExtID: uint8(midExtensionID), + RidExtID: uint8(streamIDExtensionID), + RsidExtID: uint8(repairStreamIDExtensionID), + Tries: rtxProbeCount, + OnFound: u.factory.SetStreamInfo, + }) + if !ok { + u.logger.Warnw( + "could not install stream info probe, rtx pairing will not work", nil, + "ssrc", info.SSRC, + "mime", info.MimeType, + ) } -} - -// ------------------------------------------ - -type rtxInfoReader struct { - tryTimes int - reader interceptor.RTPReader - midExtID uint8 - ridExtID uint8 - rsidExtID uint8 - factory *RTXInfoExtractorFactory - logger logger.Logger -} - -func (r *rtxInfoReader) Read(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) { - n, a, err := r.reader.Read(b, a) - if r.tryTimes < 0 || err != nil { - return n, a, err - } - - if a == nil { - a = make(interceptor.Attributes) - } - header, err := a.GetRTPHeader(b[:n]) - if err != nil { - return n, a, nil - } - - var mid, rid, rsid string - if payload := header.GetExtension(r.midExtID); payload != nil { - mid = string(payload) - } - - if payload := header.GetExtension(r.ridExtID); payload != nil { - rid = string(payload) - } - - if payload := header.GetExtension(r.rsidExtID); payload != nil { - rsid = string(payload) - } - - if mid != "" && (rid != "" || rsid != "") { - r.logger.Debugw("stream found", "mid", mid, "rid", rid, "rsid", rsid, "ssrc", header.SSRC) - r.tryTimes = -1 - go r.factory.SetStreamInfo(header.SSRC, mid, rid, rsid) - } else { - // ignore padding only packet for probe count - if !header.Padding || n-header.MarshalSize()-int(b[n-1]) != 0 { - r.tryTimes-- - } - } - return n, a, nil + + return reader } diff --git a/pkg/testutils/vnettest/vnettest.go b/pkg/testutils/vnettest/vnettest.go new file mode 100644 index 000000000..eade3314d --- /dev/null +++ b/pkg/testutils/vnettest/vnettest.go @@ -0,0 +1,231 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package vnettest sets up real pion peer connections on an in-memory virtual +// network, for integration tests that exercise media paths without a server. +// +// It depends only on pion, so it can be imported both by tests inside pkg/... and by +// the top level test package. +package vnettest + +import ( + "fmt" + "io" + "sync" + "testing" + "time" + + "github.com/pion/interceptor" + "github.com/pion/logging" + "github.com/pion/sdp/v3" + "github.com/pion/transport/v4/packetio" + "github.com/pion/transport/v4/vnet" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" +) + +const ( + VP8PayloadType = 96 + RTXPayloadType = VP8PayloadType + 1 + OpusPayloadType = 111 +) + +// Hosts are the two ends of a started virtual network. +type Hosts struct { + OfferNet *vnet.Net + AnswerNet *vnet.Net +} + +// NewHosts returns two hosts on a started virtual network, torn down with the test. +func NewHosts(t *testing.T) *Hosts { + t.Helper() + + wan, err := vnet.NewRouter(&vnet.RouterConfig{ + CIDR: "1.2.3.0/24", + LoggerFactory: logging.NewDefaultLoggerFactory(), + }) + require.NoError(t, err) + + offerNet, err := vnet.NewNet(&vnet.NetConfig{StaticIPs: []string{"1.2.3.4"}}) + require.NoError(t, err) + require.NoError(t, wan.AddNet(offerNet)) + + answerNet, err := vnet.NewNet(&vnet.NetConfig{StaticIPs: []string{"1.2.3.5"}}) + require.NoError(t, err) + require.NoError(t, wan.AddNet(answerNet)) + + require.NoError(t, wan.Start()) + t.Cleanup(func() { _ = wan.Stop() }) + + return &Hosts{OfferNet: offerNet, AnswerNet: answerNet} +} + +// NewSettingEngine returns a setting engine bound to net, with ICE timeouts short +// enough to keep tests quick. +func NewSettingEngine(net *vnet.Net) webrtc.SettingEngine { + se := webrtc.SettingEngine{} + se.SetNet(net) + se.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeUDP4}) + se.SetICETimeouts(5*time.Second, 5*time.Second, 500*time.Millisecond) + return se +} + +// MediaEngineConfig describes what to register on a media engine. +type MediaEngineConfig struct { + Video bool // VP8 and its RTX codec; otherwise opus + HeaderExtensions bool // abs-send-time + transport-cc + SimulcastExtensions bool // mid + rid + rsid +} + +func VideoRTCPFeedback() []webrtc.RTCPFeedback { + return []webrtc.RTCPFeedback{ + {Type: webrtc.TypeRTCPFBNACK}, + {Type: webrtc.TypeRTCPFBNACK, Parameter: "pli"}, + {Type: webrtc.TypeRTCPFBTransportCC}, + {Type: webrtc.TypeRTCPFBGoogREMB}, + } +} + +func NewMediaEngine(t *testing.T, cfg MediaEngineConfig) *webrtc.MediaEngine { + t.Helper() + + me := &webrtc.MediaEngine{} + kind := webrtc.RTPCodecTypeAudio + if cfg.Video { + kind = webrtc.RTPCodecTypeVideo + + require.NoError(t, me.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeVP8, ClockRate: 90000, RTCPFeedback: VideoRTCPFeedback(), + }, + PayloadType: VP8PayloadType, + }, kind)) + require.NoError(t, me.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeRTX, + ClockRate: 90000, + SDPFmtpLine: fmt.Sprintf("apt=%d", VP8PayloadType), + }, + PayloadType: RTXPayloadType, + }, kind)) + } else { + require.NoError(t, me.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2, + }, + PayloadType: OpusPayloadType, + }, kind)) + } + + if cfg.HeaderExtensions { + require.NoError(t, me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: sdp.ABSSendTimeURI}, kind)) + require.NoError(t, me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: sdp.TransportCCURI}, kind)) + } + if cfg.SimulcastExtensions { + require.NoError(t, webrtc.ConfigureSimulcastExtensionHeaders(me)) + } + + return me +} + +// PCConfig describes a peer connection on the virtual network. +type PCConfig struct { + Net *vnet.Net + MediaEngine MediaEngineConfig + + // BufferFactory is SettingEngine.BufferFactory, e. g. buffer.Factory.GetOrNew. + // Optional. + BufferFactory func(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser +} + +// NewPeerConnection builds a peer connection on the virtual network with no +// interceptors, so nothing rewrites what a test puts on the wire. +func NewPeerConnection(t *testing.T, cfg PCConfig) *webrtc.PeerConnection { + t.Helper() + + se := NewSettingEngine(cfg.Net) + se.BufferFactory = cfg.BufferFactory + + api := webrtc.NewAPI( + webrtc.WithMediaEngine(NewMediaEngine(t, cfg.MediaEngine)), + webrtc.WithSettingEngine(se), + webrtc.WithInterceptorRegistry(&interceptor.Registry{}), + ) + + pc, err := api.NewPeerConnection(webrtc.Configuration{}) + require.NoError(t, err) + t.Cleanup(func() { _ = pc.Close() }) + + return pc +} + +// GatheredOffer creates an offer and waits for gathering, so the SDP carries every +// candidate and the caller needs no trickle. +func GatheredOffer(t *testing.T, pc *webrtc.PeerConnection) webrtc.SessionDescription { + t.Helper() + + offer, err := pc.CreateOffer(nil) + require.NoError(t, err) + + gathered := webrtc.GatheringCompletePromise(pc) + require.NoError(t, pc.SetLocalDescription(offer)) + <-gathered + + return *pc.LocalDescription() +} + +// SignalPair performs a full offer/answer exchange between two peer connections and +// waits for both to connect. +func SignalPair(t *testing.T, offerer, answerer *webrtc.PeerConnection) { + t.Helper() + + connected := UntilConnected(offerer, answerer) + + require.NoError(t, answerer.SetRemoteDescription(GatheredOffer(t, offerer))) + + answer, err := answerer.CreateAnswer(nil) + require.NoError(t, err) + gathered := webrtc.GatheringCompletePromise(answerer) + require.NoError(t, answerer.SetLocalDescription(answer)) + <-gathered + + require.NoError(t, offerer.SetRemoteDescription(*answerer.LocalDescription())) + + select { + case <-connected: + case <-time.After(30 * time.Second): + t.Fatal("timed out waiting for peer connections to connect") + } +} + +// UntilConnected closes the returned channel once every peer connection is connected. +func UntilConnected(pcs ...*webrtc.PeerConnection) <-chan struct{} { + var wg sync.WaitGroup + wg.Add(len(pcs)) + for _, pc := range pcs { + var once sync.Once + pc.OnConnectionStateChange(func(s webrtc.PeerConnectionState) { + if s == webrtc.PeerConnectionStateConnected { + once.Do(wg.Done) + } + }) + } + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + return done +} diff --git a/test/integration_helpers.go b/test/integration_helpers.go index 0ac06febd..ec3469620 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -18,11 +18,14 @@ import ( "context" "fmt" "net/http" + "strings" "sync" "testing" "time" + "github.com/pion/transport/v4/vnet" "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/require" "github.com/twitchtv/twirp" "github.com/livekit/mediatransportutil/pkg/rtcconfig" @@ -33,9 +36,12 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" + "github.com/livekit/livekit-server/pkg/rtc" "github.com/livekit/livekit-server/pkg/service" + "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/livekit-server/pkg/testutils" + "github.com/livekit/livekit-server/pkg/testutils/vnettest" testclient "github.com/livekit/livekit-server/test/client" ) @@ -386,3 +392,67 @@ func stopClients(clients ...*testclient.RTCClient) { c.Stop() } } + +// ----------------------------------------------------------------------------- +// vnet media harness +// +// Setup specific to driving a real server transport over a virtual network. The +// pion side lives in pkg/testutils/vnettest, shared with the pkg/sfu media tests. +// ----------------------------------------------------------------------------- + +// newVNetWebRTCConfig builds the server side WebRTCConfig on net. The direction +// configs come from the production NewWebRTCConfig so the negotiated extensions and +// feedback stay in step with it; the setting engine is replaced so no real socket or +// ICE mux is bound. +func newVNetWebRTCConfig(t *testing.T, net *vnet.Net, bufferFactory *buffer.Factory) *rtc.WebRTCConfig { + t.Helper() + + conf, err := config.NewConfig("", true, nil, nil) + require.NoError(t, err) + + // an ephemeral port range instead of the dev mode single port, which would bind a mux + conf.RTC.TCPPort = 0 + conf.RTC.UDPPort = rtcconfig.PortRange{} + conf.RTC.ICEPortRangeStart = 50000 + conf.RTC.ICEPortRangeEnd = 60000 + + rtcConf, err := rtc.NewWebRTCConfig(conf) + require.NoError(t, err) + require.Nil(t, rtcConf.UDPMux, "test config must not bind a udp mux") + + rtcConf.SettingEngine = vnettest.NewSettingEngine(net) + rtcConf.SetBufferFactory(bufferFactory) + + return rtcConf +} + +// stripDeclaredSSRCs removes the a=ssrc lines pion puts in its offer. Browsers doing +// rid based simulcast do not declare per-layer SSRCs, which is why a repair SSRC has to +// be learned at all; leaving them in would let the receiver resolve everything from SDP. +func stripDeclaredSSRCs(offer string) string { + lines := strings.Split(offer, "\r\n") + filtered := lines[:0] + for _, line := range lines { + if strings.HasPrefix(line, "a=ssrc") { + continue + } + filtered = append(filtered, line) + } + return strings.Join(filtered, "\r\n") +} + +// sendUntil calls send every 20ms until done reports true or the timeout expires, +// returning the final state of done. +func sendUntil(t *testing.T, timeout time.Duration, done func() bool, send func()) bool { + t.Helper() + + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if done() { + return true + } + send() + time.Sleep(20 * time.Millisecond) + } + return done() +} diff --git a/test/rtx_pairing_integration_test.go b/test/rtx_pairing_integration_test.go new file mode 100644 index 000000000..7a497bb8e --- /dev/null +++ b/test/rtx_pairing_integration_test.go @@ -0,0 +1,562 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package test + +// RTX repair stream pairing on simulcast (rid) streams, over a real PCTransport driven +// by a real pion publisher on a virtual network. +// +// RID based simulcast has no a=ssrc-group:FID line, so the pairing cannot come from +// SDP. The repair SSRC of a layer comes either from the mid/rsid header extensions on +// its packets, or - for a migrated publisher, which is mid-stream and no longer sends +// those extensions - from the migration info in TransportParams.SimTracks. +// +// Neither source fails loudly when it breaks: the repair buffer just accumulates +// packets that are never applied and NACK recovery for simulcast stops working. Both +// paths are therefore asserted end to end, by retransmitting a sequence number that is +// never sent on the primary stream and requiring it to surface on the primary buffer. + +import ( + "encoding/binary" + "fmt" + "sync" + "testing" + "time" + + "github.com/pion/rtp" + "github.com/pion/sdp/v3" + "github.com/pion/transport/v4/vnet" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/rtc" + "github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes" + "github.com/livekit/livekit-server/pkg/sfu/buffer" + sfuinterceptor "github.com/livekit/livekit-server/pkg/sfu/interceptor" + "github.com/livekit/livekit-server/pkg/testutils/vnettest" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +// the single video m-line of the publisher's offer +const rtxTestMid = "0" + +const ( + sendsExtensions = false + omitsExtensions = true +) + +// SSRCs are chosen by the test rather than taken from the offer; see stripDeclaredSSRCs. +var rtxTestLayers = []struct { + rid string + ssrc uint32 + rtxSSRC uint32 + recoveredSeq uint16 +}{ + {rid: "q", ssrc: 1001, rtxSSRC: 2001, recoveredSeq: 50001}, + {rid: "h", ssrc: 1002, rtxSSRC: 2002, recoveredSeq: 50002}, + {rid: "f", ssrc: 1003, rtxSSRC: 2003, recoveredSeq: 50003}, +} + +// TestSimulcastRTXPairing covers a publisher sending mid/rid/rsid: the pairing comes +// from probing the packets. +func TestSimulcastRTXPairing(t *testing.T) { + h := newRTXHarness(t, nil) + + h.run(t, sendsExtensions) + + // pairing is also reported through the callback mediatrack subscribes to + require.Equal(t, len(rtxTestLayers), h.tracker.pairCount(), "not all rtx pairs found: %s", h.tracker.describe()) + for _, w := range h.writers { + base, repair := h.tracker.pair(w.rid) + require.Equal(t, w.ssrc, base, "wrong base ssrc paired for rid %q", w.rid) + require.Equal(t, w.rtxSSRC, repair, "wrong repair ssrc paired for rid %q", w.rid) + } +} + +// TestSimulcastRTXPairingAfterMigration covers a migrated publisher: it is mid-stream +// and sends no mid/rid/rsid, so UnhandleSimulcastInterceptor synthesises them for pion +// and the pairing has to come from SimTracks. RepairSSRC names the repair stream. +func TestSimulcastRTXPairingAfterMigration(t *testing.T) { + simTracks := make(map[uint32]sfuinterceptor.SimulcastTrackInfo, 2*len(rtxTestLayers)) + for _, l := range rtxTestLayers { + simTracks[l.ssrc] = sfuinterceptor.SimulcastTrackInfo{ + Mid: rtxTestMid, + StreamID: l.rid, + RepairSSRC: l.rtxSSRC, + } + simTracks[l.rtxSSRC] = sfuinterceptor.SimulcastTrackInfo{ + Mid: rtxTestMid, + StreamID: l.rid, + IsRepairStream: true, + } + } + + newRTXHarness(t, simTracks).run(t, omitsExtensions) +} + +// TestSimulcastRTXPairingAfterMigrationWithoutRepairSSRC covers migration info that +// marks the repair stream but leaves RepairSSRC unset on the primary entry. +func TestSimulcastRTXPairingAfterMigrationWithoutRepairSSRC(t *testing.T) { + simTracks := make(map[uint32]sfuinterceptor.SimulcastTrackInfo, 2*len(rtxTestLayers)) + for _, l := range rtxTestLayers { + simTracks[l.ssrc] = sfuinterceptor.SimulcastTrackInfo{Mid: rtxTestMid, StreamID: l.rid} + simTracks[l.rtxSSRC] = sfuinterceptor.SimulcastTrackInfo{ + Mid: rtxTestMid, + StreamID: l.rid, + IsRepairStream: true, + } + } + + newRTXHarness(t, simTracks).run(t, omitsExtensions) +} + +// ----------------------------------------------------------------------------- +// harness +// ----------------------------------------------------------------------------- + +type rtxHarness struct { + transport *rtc.PCTransport + pubPC *webrtc.PeerConnection + writers []*simulcastWriter + tracker *rtxPairTracker +} + +func newRTXHarness(t *testing.T, simTracks map[uint32]sfuinterceptor.SimulcastTrackInfo) *rtxHarness { + t.Helper() + + hosts := vnettest.NewHosts(t) + tracker := newRTXPairTracker() + + bufferFactory := buffer.NewFactoryOfBufferFactory(500, 200).CreateBufferFactory() + pcTransport := newPublisherTransportForTest(t, hosts.AnswerNet, bufferFactory, simTracks, tracker) + pubPC, writers := newSimulcastPublisherPC(t, hosts.OfferNet) + + return &rtxHarness{ + transport: pcTransport, + pubPC: pubPC, + writers: writers, + tracker: tracker, + } +} + +// run negotiates, publishes every layer, then retransmits a sequence number that is +// never sent on the primary stream and waits for it to surface on the primary buffer. +func (h *rtxHarness) run(t *testing.T, omitExtensions bool) { + t.Helper() + + signalToTransport(t, h.pubPC, h.transport) + require.Equal(t, rtxTestMid, h.pubPC.GetTransceivers()[0].Mid()) + for _, w := range h.writers { + w.mid = rtxTestMid + w.omitExtensions = omitExtensions + } + + // every layer has to bind before RTX is sent, which is also the production ordering: + // a retransmission only follows a NACK for an established layer + require.True( + t, + sendUntil(t, 20*time.Second, func() bool { return h.tracker.boundCount() == len(rtxTestLayers) }, func() { + for _, w := range h.writers { + w.writePrimary(t) + } + }), + "timed out waiting for all simulcast layers to bind: %s", h.tracker.describe(), + ) + + require.True( + t, + sendUntil(t, 20*time.Second, func() bool { + for _, w := range h.writers { + if !h.tracker.sawSeq(w.rid, w.recoveredSeq) { + return false + } + } + return true + }, func() { + for _, w := range h.writers { + w.writeRepair(t, h.tracker.rtxPayloadType(w.rid), w.recoveredSeq) + } + }), + "retransmissions never recovered into the primary buffers: %s", h.tracker.describe(), + ) +} + +// newPublisherTransportForTest builds the production publisher transport on net and +// wires up what ParticipantImpl/MediaTrack do with a published layer. +func newPublisherTransportForTest( + t *testing.T, + net *vnet.Net, + bufferFactory *buffer.Factory, + simTracks map[uint32]sfuinterceptor.SimulcastTrackInfo, + tracker *rtxPairTracker, +) *rtc.PCTransport { + t.Helper() + + rtcConf := newVNetWebRTCConfig(t, net, bufferFactory) + + handler := &transportfakes.FakeHandler{} + params := rtc.TransportParams{ + Handler: handler, + Config: rtcConf, + DirectionConfig: rtcConf.Publisher, + ProtocolVersion: 6, + Logger: logger.GetLogger(), + Transport: livekit.SignalTarget_PUBLISHER, + SimTracks: simTracks, + EnabledPublishCodecs: []*livekit.Codec{ + {Mime: webrtc.MimeTypeVP8}, + {Mime: webrtc.MimeTypeRTX}, + }, + // all candidates are carried in the answer, so the test needs no trickle + UseOneShotSignallingMode: true, + } + + pcTransport, err := rtc.NewPCTransport(params) + require.NoError(t, err) + t.Cleanup(pcTransport.Close) + + // mirror mediatrack.addReceiver: bind the buffer of each published layer, subscribe + // to the pairing notification, and drain the buffer the way WebRTCReceiver does + handler.OnTrackCalls(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) { + rid, ssrc := track.RID(), uint32(track.SSRC()) + + buff := bufferFactory.GetBuffer(ssrc) + if buff == nil { + t.Errorf("no buffer for published ssrc %d (rid %q)", ssrc, rid) + return + } + if err := buff.Bind(receiver.GetParameters(), track.Codec().RTPCodecCapability, 0); err != nil { + t.Errorf("binding buffer for rid %q failed: %v", rid, err) + return + } + buff.OnNotifyRTX(func(base, repair uint32, rsid string) { + tracker.pairFound(rsid, base, repair) + }) + + // mirror ParticipantImpl.onMediaTrack + pcTransport.RTPStreamPublished(ssrc, pcTransport.GetMid(receiver), rid) + + tracker.layerBound(rid, ssrc, buff, receiver.GetParameters()) + go tracker.drain(rid, buff) + }) + + return pcTransport +} + +// signalToTransport runs a one-shot offer/answer against the transport and waits for +// the publisher to connect. +func signalToTransport(t *testing.T, pub *webrtc.PeerConnection, pcTransport *rtc.PCTransport) { + t.Helper() + + connected := vnettest.UntilConnected(pub) + + offer := vnettest.GatheredOffer(t, pub) + offer.SDP = stripDeclaredSSRCs(offer.SDP) + require.NoError(t, pcTransport.HandleRemoteDescription(offer, 1)) + + answer, _, err := pcTransport.GetAnswer() + require.NoError(t, err) + require.NoError(t, pub.SetRemoteDescription(answer)) + + select { + case <-connected: + case <-time.After(30 * time.Second): + t.Fatal("timed out waiting for the publisher to connect") + } +} + +// ----------------------------------------------------------------------------- +// publisher: raw simulcast writer with a per-layer repair stream +// ----------------------------------------------------------------------------- + +// rawTrackLocal hands the test the negotiated write stream directly. Unlike +// TrackLocalStaticRTP it does not rewrite SSRC or payload type, which is what lets a +// repair stream be emitted on its own SSRC alongside the primary stream of the same rid. +type rawTrackLocal struct { + id string + streamID string + rid string + + lock sync.Mutex + writers []webrtc.TrackLocalWriter + exts []webrtc.RTPHeaderExtensionParameter +} + +func (t *rawTrackLocal) Bind(ctx webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, error) { + for _, c := range ctx.CodecParameters() { + if c.PayloadType != vnettest.VP8PayloadType { + continue + } + + t.lock.Lock() + t.writers = append(t.writers, ctx.WriteStream()) + t.exts = ctx.HeaderExtensions() + t.lock.Unlock() + return c, nil + } + return webrtc.RTPCodecParameters{}, fmt.Errorf("vp8 not negotiated for rid %q", t.rid) +} + +func (t *rawTrackLocal) Unbind(webrtc.TrackLocalContext) error { return nil } +func (t *rawTrackLocal) ID() string { return t.id } +func (t *rawTrackLocal) RID() string { return t.rid } +func (t *rawTrackLocal) StreamID() string { return t.streamID } +func (t *rawTrackLocal) Kind() webrtc.RTPCodecType { return webrtc.RTPCodecTypeVideo } + +func (t *rawTrackLocal) extensionID(uri string) uint8 { + t.lock.Lock() + defer t.lock.Unlock() + + for _, e := range t.exts { + if e.URI == uri { + return uint8(e.ID) + } + } + return 0 +} + +func (t *rawTrackLocal) write(header *rtp.Header, payload []byte) { + t.lock.Lock() + writers := append([]webrtc.TrackLocalWriter(nil), t.writers...) + t.lock.Unlock() + + for _, w := range writers { + _, _ = w.WriteRTP(header, payload) + } +} + +// simulcastWriter emits the primary and repair streams of one simulcast layer. +type simulcastWriter struct { + track *rawTrackLocal + mid string + rid string + ssrc uint32 + rtxSSRC uint32 + + // omitExtensions emulates a migrated publisher, which sends no mid/rid/rsid + omitExtensions bool + + // recoveredSeq is only ever sent inside an RTX payload, never on the primary + // stream, so its arrival on the primary buffer proves RTX recovery worked + recoveredSeq uint16 + + lock sync.Mutex + seq uint16 +} + +func (w *simulcastWriter) nextSeq() uint16 { + w.lock.Lock() + defer w.lock.Unlock() + + w.seq++ + return w.seq +} + +func (w *simulcastWriter) header(t *testing.T, ssrc uint32, pt uint8, seq uint16, rid, rsid string) *rtp.Header { + t.Helper() + + h := &rtp.Header{ + Version: 2, + PayloadType: pt, + SequenceNumber: seq, + Timestamp: uint32(seq) * 3000, + SSRC: ssrc, + } + if w.omitExtensions { + return h + } + + midID := w.track.extensionID(sdp.SDESMidURI) + require.NotZero(t, midID, "sdes:mid not negotiated") + require.NoError(t, h.SetExtension(midID, []byte(w.mid))) + + if rid != "" { + ridID := w.track.extensionID(sdp.SDESRTPStreamIDURI) + require.NotZero(t, ridID, "sdes:rtp-stream-id not negotiated") + require.NoError(t, h.SetExtension(ridID, []byte(rid))) + } + if rsid != "" { + rsidID := w.track.extensionID(sdp.SDESRepairRTPStreamIDURI) + require.NotZero(t, rsidID, "sdes:repaired-rtp-stream-id not negotiated") + require.NoError(t, h.SetExtension(rsidID, []byte(rsid))) + } + return h +} + +func (w *simulcastWriter) writePrimary(t *testing.T) { + t.Helper() + + w.track.write(w.header(t, w.ssrc, vnettest.VP8PayloadType, w.nextSeq(), w.rid, ""), vp8TestPayload()) +} + +// writeRepair emits an RFC 4588 repair packet: the sequence number being retransmitted +// is prepended to the payload. +func (w *simulcastWriter) writeRepair(t *testing.T, rtxPT uint8, originalSeq uint16) { + t.Helper() + + if rtxPT == 0 { + rtxPT = vnettest.RTXPayloadType + } + + inner := vp8TestPayload() + payload := make([]byte, 2+len(inner)) + binary.BigEndian.PutUint16(payload[:2], originalSeq) + copy(payload[2:], inner) + + w.track.write(w.header(t, w.rtxSSRC, rtxPT, w.nextSeq(), "", w.rid), payload) +} + +func vp8TestPayload() []byte { + return []byte{0x10, 0x00, 0x00, 0x9d, 0x01, 0x2a, 0x40, 0x01, 0xf0, 0x00} +} + +// newSimulcastPublisherPC builds the publishing peer connection and one writer per +// simulcast layer. +func newSimulcastPublisherPC(t *testing.T, net *vnet.Net) (*webrtc.PeerConnection, []*simulcastWriter) { + t.Helper() + + pc := vnettest.NewPeerConnection(t, vnettest.PCConfig{ + Net: net, + MediaEngine: vnettest.MediaEngineConfig{ + Video: true, + HeaderExtensions: true, + SimulcastExtensions: true, + }, + }) + + writers := make([]*simulcastWriter, 0, len(rtxTestLayers)) + for _, l := range rtxTestLayers { + writers = append(writers, &simulcastWriter{ + track: &rawTrackLocal{id: "video", streamID: "pion", rid: l.rid}, + rid: l.rid, + ssrc: l.ssrc, + rtxSSRC: l.rtxSSRC, + recoveredSeq: l.recoveredSeq, + }) + } + + sender, err := pc.AddTrack(writers[0].track) + require.NoError(t, err) + for _, w := range writers[1:] { + require.NoError(t, sender.AddEncoding(w.track)) + } + + return pc, writers +} + +// ----------------------------------------------------------------------------- +// tracking +// ----------------------------------------------------------------------------- + +type rtxPairTracker struct { + lock sync.Mutex + + bound map[string]uint32 // rid -> base ssrc + params map[string]webrtc.RTPParameters + pairs map[string][2]uint32 // rsid -> {base ssrc, repair ssrc} + seen map[string]map[uint16]bool // rid -> sequence numbers read off the primary buffer +} + +func newRTXPairTracker() *rtxPairTracker { + return &rtxPairTracker{ + bound: make(map[string]uint32), + params: make(map[string]webrtc.RTPParameters), + pairs: make(map[string][2]uint32), + seen: make(map[string]map[uint16]bool), + } +} + +func (t *rtxPairTracker) layerBound(rid string, ssrc uint32, buff *buffer.Buffer, params webrtc.RTPParameters) { + t.lock.Lock() + defer t.lock.Unlock() + + t.bound[rid] = ssrc + t.params[rid] = params + t.seen[rid] = make(map[uint16]bool) +} + +func (t *rtxPairTracker) pairFound(rsid string, base, repair uint32) { + t.lock.Lock() + defer t.lock.Unlock() + + t.pairs[rsid] = [2]uint32{base, repair} +} + +func (t *rtxPairTracker) boundCount() int { + t.lock.Lock() + defer t.lock.Unlock() + + return len(t.bound) +} + +func (t *rtxPairTracker) pairCount() int { + t.lock.Lock() + defer t.lock.Unlock() + + return len(t.pairs) +} + +func (t *rtxPairTracker) pair(rid string) (uint32, uint32) { + t.lock.Lock() + defer t.lock.Unlock() + + p := t.pairs[rid] + return p[0], p[1] +} + +func (t *rtxPairTracker) rtxPayloadType(rid string) uint8 { + t.lock.Lock() + defer t.lock.Unlock() + + for _, c := range t.params[rid].Codecs { + if c.MimeType == webrtc.MimeTypeRTX { + return uint8(c.PayloadType) + } + } + return 0 +} + +func (t *rtxPairTracker) sawSeq(rid string, seq uint16) bool { + t.lock.Lock() + defer t.lock.Unlock() + + return t.seen[rid][seq] +} + +// drain consumes the primary buffer the way WebRTCReceiver does, recording which +// sequence numbers made it through. +func (t *rtxPairTracker) drain(rid string, buff *buffer.Buffer) { + b := make([]byte, 1500) + for { + ep, err := buff.ReadExtended(b) + if err != nil { + return + } + if ep == nil || ep.Packet == nil { + continue + } + + t.lock.Lock() + t.seen[rid][ep.Packet.SequenceNumber] = true + t.lock.Unlock() + } +} + +func (t *rtxPairTracker) describe() string { + t.lock.Lock() + defer t.lock.Unlock() + + return fmt.Sprintf("bound=%v pairs=%v", t.bound, t.pairs) +}