Fix simulcast RTX pairing broken by pion/webrtc#3470 (#4800)

pion no longer starts the repair stream reader when a custom BufferFactory
is set, so the mid/rid/rsid extensions were never observed and simulcast RTX
streams were never paired with their primary streams.

Extract the extensions on the buffer write path instead. Migrated publishers
send no extensions at all, so pair those from SimTracks.

Adds an integration test covering both paths, and moves the vnet setup it
shares with the downtrack test into pkg/testutils/vnettest.

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

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

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

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]

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

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
 }

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

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

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