rtc: parse each remote offer once per negotiation (#4904)

This commit is contained in:
Raja Subramanian authored and GitHub committed 2026-09-25 05:01:30 +05:30
1 parent 6b2e3ece69
commit f52b0e2bd5
7 files changed
+235 -60

No files matched your search

+1 -1
View File
@@ -161,7 +161,7 @@ tool (
golang.org/x/tools/cmd/goimports
)
replace github.com/pion/webrtc/v4 => github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1
replace github.com/pion/webrtc/v4 => github.com/livekit/webrtc-pion/v4 v4.2.18-warp.2
replace github.com/pion/dtls/v3 => github.com/livekit/dtls/v3 v3.1.5-warp.1
+2 -2
View File
@@ -168,8 +168,8 @@ github.com/livekit/protocol v1.52.1-0.20260920231100-26af364d0319 h1:mEfaX/HY18L
github.com/livekit/protocol v1.52.1-0.20260920231100-26af364d0319/go.mod h1:Ke1q64xQzwZ1S1ZH66SudXHsaaQQAFoX+TL1iLpu7XU=
github.com/livekit/psrpc v0.8.0 h1:uo3x+DL8pK2OIaCxg/s9sLUO2cIocS5HwdPhhI+m+fQ=
github.com/livekit/psrpc v0.8.0/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q=
github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1 h1:fH+v4W+NFp9FfPzON6FaUFNmazGcctaAhb2P+Ksf+1s=
github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1/go.mod h1:rbKGHo2OpNUImWTvRIV776/3xjjq/t47H3IZiTtwluc=
github.com/livekit/webrtc-pion/v4 v4.2.18-warp.2 h1:EKmLnuIs/2wRJmfQKv8JbZIGjHt7PQZYf94TAGyZGSI=
github.com/livekit/webrtc-pion/v4 v4.2.18-warp.2/go.mod h1:Ps5ynXt7Heejsbg8fclYIDCBsUC0nCHNoWRUwFqyA2k=
github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4=
github.com/mackerelio/go-osstat v0.2.8/go.mod h1:SyS3XxKdoSKJnTGTkN5Yrh6VUQVuAURACfE6y+2DN4k=
github.com/magefile/mage v1.17.2 h1:fyXVu1eadI8Ap1HCCNgEhJ5McIWiYhLR8uol64ZZc40=
+8 -14
View File
@@ -1215,20 +1215,19 @@ func (p *ParticipantImpl) HandleOffer(sd *livekit.SessionDescription) error {
lgr.Debugw("received offer")
parsedOffer, err := offer.Unmarshal()
if err != nil {
lgr.Warnw("could not parse offer", err)
return err
}
if p.params.UseOneShotSignallingMode {
parsedOffer, err := offer.Unmarshal()
if err != nil {
lgr.Warnw("could not parse offer", err)
return err
}
if err := p.synthesizeAddTrackRequests(parsedOffer); err != nil {
lgr.Warnw("could not synthesize add track requests", err)
return err
}
}
err = p.TransportManager.HandleOffer(offer, offerId, p.MigrateState() == types.MigrateStateInit)
err := p.TransportManager.HandleOffer(offer, offerId, p.MigrateState() == types.MigrateStateInit)
if err != nil {
lgr.Warnw("could not handle offer", err, "mungedOffer", offer)
return err
@@ -1243,13 +1242,8 @@ func (p *ParticipantImpl) HandleOffer(sd *livekit.SessionDescription) error {
}
func (p *ParticipantImpl) onPublisherSetRemoteDescription() {
offer := p.TransportManager.LastPublisherOfferPending()
if offer == nil {
return
}
parsedOffer, err := offer.Unmarshal()
if err != nil {
p.pubLogger.Warnw("could not parse offer", err)
parsedOffer := p.TransportManager.LastPublisherOfferParsed()
if parsedOffer == nil {
return
}
+12 -6
View File
@@ -319,9 +319,8 @@ func (p *ParticipantImpl) setCodecPreferencesOpusRedForPublisher(
// configure publisher answer for audio track's dtx and stereo settings
func (p *ParticipantImpl) configurePublisherAnswer(answer webrtc.SessionDescription) webrtc.SessionDescription {
offer := p.TransportManager.LastPublisherOffer()
parsedOffer, err := offer.Unmarshal()
if err != nil {
parsedOffer := p.TransportManager.LastPublisherOfferParsed()
if parsedOffer == nil {
return answer
}
@@ -330,6 +329,9 @@ func (p *ParticipantImpl) configurePublisherAnswer(answer webrtc.SessionDescript
return answer
}
// payload type lookup scans the whole answer, resolve it once
opusPT, opusPTErr := parsedAnswer.GetPayloadTypeForCodec(sdp.Codec{Name: mime.MimeTypeCodecOpus.String()})
changed := false
for _, m := range parsedAnswer.MediaDescriptions {
switch m.MediaName.Media {
case "audio":
@@ -371,9 +373,8 @@ func (p *ParticipantImpl) configurePublisherAnswer(answer webrtc.SessionDescript
continue
}
opusPT, err := parsedAnswer.GetPayloadTypeForCodec(sdp.Codec{Name: mime.MimeTypeCodecOpus.String()})
if err != nil {
p.pubLogger.Infow("failed to get opus payload type", "error", err, "trackID", ti.Sid)
if opusPTErr != nil {
p.pubLogger.Infow("failed to get opus payload type", "error", opusPTErr, "trackID", ti.Sid)
continue
}
@@ -390,6 +391,7 @@ func (p *ParticipantImpl) configurePublisherAnswer(answer webrtc.SessionDescript
attr.Value = strings.ReplaceAll(attr.Value, ";stereo=1", "")
}
m.Attributes[i] = attr
changed = true
}
}
@@ -398,6 +400,10 @@ func (p *ParticipantImpl) configurePublisherAnswer(answer webrtc.SessionDescript
}
}
if !changed {
return answer
}
bytes, err := parsedAnswer.Marshal()
if err != nil {
p.pubLogger.Infow("failed to marshal answer", "error", err)
+65 -31
View File
@@ -300,6 +300,9 @@ type PCTransport struct {
signalStateCheckTimer *time.Timer
currentOfferIceCredential string // ice user:pwd, for publish side ice restart checking
pendingRestartIceOffer *webrtc.SessionDescription
// parsed forms kept so that a remote description is parsed once per negotiation
pendingRestartIceOfferParsed *sdp.SessionDescription
remoteOfferParsed *sdp.SessionDescription
}
type TransportParams struct {
@@ -1646,6 +1649,7 @@ func (t *PCTransport) clearConnTimer() {
}
func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remoteId uint32) error {
parsed, parseErr := sd.Unmarshal()
if t.params.UseOneShotSignallingMode {
if sd.Type == webrtc.SDPTypeOffer {
remoteOfferId := t.remoteOfferId.Load()
@@ -1666,8 +1670,7 @@ func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remo
}
// add remote candidates to ICE connection details
parsed, err := sd.Unmarshal()
if err == nil {
if parseErr == nil {
addRemoteICECandidates := func(attrs []sdp.Attribute) {
for _, a := range attrs {
if a.IsICECandidate() {
@@ -1686,11 +1689,11 @@ func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remo
}
}
err = t.pc.SetRemoteDescription(sd)
if err != nil {
if err := t.pc.SetRemoteDescription(sd); err != nil {
t.params.Logger.Errorw("could not set remote description on synchronous mode peer connection", err)
return err
}
t.setRemoteOfferParsed(sd.Type, parsed)
rtxRepairs := nonSimulcastRTXRepairsFromSDP(parsed, t.params.Logger)
if len(rtxRepairs) > 0 {
@@ -1706,6 +1709,8 @@ func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remo
signal: signalRemoteDescriptionReceived,
data: remoteDescriptionData{
sessionDescription: &sd,
parsed: parsed,
parseErr: parseErr,
remoteId: remoteId,
},
})
@@ -1907,6 +1912,7 @@ func (t *PCTransport) HandleICERestartSDPFragment(sdpFragment string) (string, e
t.params.Logger.Warnw("could not set remote description", err)
return "", err
}
t.setRemoteOfferParsed(sd.Type, parsedRemote)
// clear out connection details on ICE restart and re-populate
t.connectionDetails.Clear()
@@ -2289,6 +2295,23 @@ func (t *PCTransport) SetPreviousSdp(localDescription, remoteDescription *webrtc
t.lock.Unlock()
}
func (t *PCTransport) setRemoteOfferParsed(sdType webrtc.SDPType, parsed *sdp.SessionDescription) {
if sdType != webrtc.SDPTypeOffer {
return
}
t.lock.Lock()
t.remoteOfferParsed = parsed
t.lock.Unlock()
}
// RemoteOfferParsed returns the parsed form of the last remote offer set on the peer connection,
// candidates filtered the same way pion saw them
func (t *PCTransport) RemoteOfferParsed() *sdp.SessionDescription {
t.lock.RLock()
defer t.lock.RUnlock()
return t.remoteOfferParsed
}
func (t *PCTransport) parseTrackMid(sd webrtc.SessionDescription, senders map[string]*webrtc.RTPSender) error {
parsed, err := sd.Unmarshal()
if err != nil {
@@ -2369,22 +2392,17 @@ func (t *PCTransport) handleICEGatheringCompleteAnswerer() error {
}
offer := *t.pendingRestartIceOffer
parsed := t.pendingRestartIceOfferParsed
t.pendingRestartIceOffer = nil
t.pendingRestartIceOfferParsed = nil
t.params.Logger.Debugw("accept remote restart ice offer after ICE gathering")
// Parse the offer payload types before SetRemoteDescription so this does not
// race with pion's use of the same description.
var offerAudioPT map[mime.MimeType]webrtc.PayloadType
if parsed, err := offer.Unmarshal(); err == nil {
offerAudioPT = offerAudioPayloadTypes(parsed)
}
if err := t.setRemoteDescription(offer); err != nil {
if err := t.setRemoteDescription(offer, parsed); err != nil {
return err
}
t.params.Handler.OnSetRemoteDescriptionOffer()
t.processSendersPendingConfig(offerAudioPT)
t.processSendersPendingConfig(offerAudioPayloadTypes(parsed))
return t.createAndSendAnswer()
}
@@ -2498,10 +2516,23 @@ func (t *PCTransport) isCandidateFilterActive(preferTCP bool) bool {
}
func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, isLocal bool) webrtc.SessionDescription {
parsed, err := sd.Unmarshal()
if err != nil {
t.params.Logger.Warnw("could not unmarshal SDP to filter candidates", err)
return sd
return t.filterCandidatesParsed(sd, nil, preferTCP, isLocal)
}
// filterCandidatesParsed filters using an already parsed description when available,
// the parsed description is modified in place to match the returned SDP
func (t *PCTransport) filterCandidatesParsed(
sd webrtc.SessionDescription,
parsed *sdp.SessionDescription,
preferTCP, isLocal bool,
) webrtc.SessionDescription {
if parsed == nil {
var err error
parsed, err = sd.Unmarshal()
if err != nil {
t.params.Logger.Warnw("could not unmarshal SDP to filter candidates", err)
return sd
}
}
filterAttributes := func(attrs []sdp.Attribute) []sdp.Attribute {
@@ -2727,15 +2758,20 @@ func (t *PCTransport) handleSendOffer(_ event) error {
type remoteDescriptionData struct {
sessionDescription *webrtc.SessionDescription
parsed *sdp.SessionDescription
parseErr error // set when parsed is nil
remoteId uint32
}
func (t *PCTransport) handleRemoteDescriptionReceived(e event) error {
rdd := e.data.(remoteDescriptionData)
if rdd.parsed == nil {
return rdd.parseErr
}
if rdd.sessionDescription.Type == webrtc.SDPTypeOffer {
return t.handleRemoteOfferReceived(rdd.sessionDescription, rdd.remoteId)
return t.handleRemoteOfferReceived(rdd.sessionDescription, rdd.parsed, rdd.remoteId)
} else {
return t.handleRemoteAnswerReceived(rdd.sessionDescription, rdd.remoteId)
return t.handleRemoteAnswerReceived(rdd.sessionDescription, rdd.parsed, rdd.remoteId)
}
}
@@ -2751,13 +2787,13 @@ func (t *PCTransport) isRemoteOfferRestartICE(parsed *sdp.SessionDescription) (s
return credential, restartICE, nil
}
func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error {
func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription, parsed *sdp.SessionDescription) error {
// filter before setting remote description so that pion does not see filtered remote candidates
preferTCP := t.preferTCP.Load()
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("remote description (unfiltered)", "type", sd.Type, "sdp", sd.SDP)
}
sd = t.filterCandidates(sd, preferTCP, false)
sd = t.filterCandidatesParsed(sd, parsed, preferTCP, false)
if t.isCandidateFilterActive(preferTCP) {
t.params.Logger.Debugw("remote description (filtered)", "type", sd.Type, "sdp", sd.SDP)
}
@@ -2774,7 +2810,9 @@ func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error {
}
prometheus.RecordServiceOperationError(sdpType, "remote_description")
return errors.Wrap(err, "setting remote description failed")
} else if sd.Type == webrtc.SDPTypeAnswer {
}
t.setRemoteOfferParsed(sd.Type, parsed)
if sd.Type == webrtc.SDPTypeAnswer {
t.lock.Lock()
if !t.canReuseTransceiver {
t.canReuseTransceiver = true
@@ -2871,7 +2909,7 @@ func (t *PCTransport) createAndSendAnswer() error {
return t.localDescriptionSent()
}
func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, offerId uint32) error {
func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, parsed *sdp.SessionDescription, offerId uint32) error {
t.params.Logger.Debugw("processing offer", "offerId", offerId)
remoteOfferId := t.remoteOfferId.Load()
if remoteOfferId != 0 && remoteOfferId != t.localAnswerId.Load() {
@@ -2884,11 +2922,6 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, o
}
t.remoteOfferId.Store(offerId)
parsed, err := sd.Unmarshal()
if err != nil {
return err
}
t.lock.Lock()
if !t.firstOfferReceived {
t.firstOfferReceived = true
@@ -2915,6 +2948,7 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, o
if offerRestartICE && t.pc.ICEGatheringState() == webrtc.ICEGatheringStateGathering {
t.params.Logger.Debugw("remote offer restart ice while ice gathering")
t.pendingRestartIceOffer = sd
t.pendingRestartIceOfferParsed = parsed
return nil
}
@@ -2928,7 +2962,7 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, o
isStartOfConnectionSequence := t.pc.RemoteDescription() == nil
if err := t.setRemoteDescription(*sd); err != nil {
if err := t.setRemoteDescription(*sd, parsed); err != nil {
return err
}
@@ -2954,7 +2988,7 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, o
return t.createAndSendAnswer()
}
func (t *PCTransport) handleRemoteAnswerReceived(sd *webrtc.SessionDescription, answerId uint32) error {
func (t *PCTransport) handleRemoteAnswerReceived(sd *webrtc.SessionDescription, parsed *sdp.SessionDescription, answerId uint32) error {
t.params.Logger.Debugw("processing answer", "answerId", answerId)
if answerId != 0 && answerId != t.localOfferId.Load() {
t.params.Logger.Warnw(
@@ -2967,7 +3001,7 @@ func (t *PCTransport) handleRemoteAnswerReceived(sd *webrtc.SessionDescription,
t.clearSignalStateCheckTimer()
if err := t.setRemoteDescription(*sd); err != nil {
if err := t.setRemoteDescription(*sd, parsed); err != nil {
// Pion will call RTPSender.Send method for each new added Downtrack, and return error if the DownTrack.Bind
// returns error. In case of Downtrack.Bind returns ErrUnsupportedCodec, the signal state will be stable as negotiation is already completed
// before startRTPSenders, and the peerconnection state can be recovered by next negotiation which will be triggered
+144
View File
@@ -28,9 +28,11 @@ import (
"github.com/livekit/livekit-server/pkg/rtc/transport"
"github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/livekit-server/pkg/testutils"
"github.com/livekit/protocol/codecs/mime"
"github.com/livekit/protocol/livekit"
lksdp "github.com/livekit/protocol/sdp"
)
func TestMissingAnswerDuringICERestart(t *testing.T) {
@@ -803,3 +805,145 @@ func TestSinglePCAnswerStripsSubscribeOnlyCodecsFromRecvSide(t *testing.T) {
"answer must not advertise H.264 in recv-side m-section: %s", a.Value)
}
}
// the parsed remote offer kept by the transport must follow every accepted offer
// and survive an offer that does not parse
func TestRemoteOfferParsedTracksOffers(t *testing.T) {
codecs := []*livekit.Codec{{Mime: mime.MimeTypeVP8.String()}}
handler := &transportfakes.FakeHandler{}
tr, err := NewPCTransport(TransportParams{
Config: &WebRTCConfig{
// offers from a real client carry RTX pairs that the transport registers here
BufferFactory: buffer.NewFactoryOfBufferFactory(500, 200).CreateBufferFactory(),
},
EnabledPublishCodecs: codecs,
IsOfferer: false,
Handler: handler,
})
require.NoError(t, err)
defer tr.Close()
var clientME webrtc.MediaEngine
require.NoError(t, registerCodecs(&clientME, codecs, RTCPFeedbackConfig{}, false))
client, err := webrtc.NewAPI(webrtc.WithMediaEngine(&clientME)).NewPeerConnection(webrtc.Configuration{})
require.NoError(t, err)
defer client.Close()
handler.OnAnswerCalls(func(sd webrtc.SessionDescription, _ uint32, _ map[string]string) error {
return client.SetRemoteDescription(sd)
})
mids := func(parsed *sdp.SessionDescription) []string {
var out []string
for _, m := range parsed.MediaDescriptions {
mid, _ := m.Attribute(sdp.AttrKeyMID)
out = append(out, mid)
}
return out
}
var offerId uint32
sendOffer := func(numSections int) {
_, err := client.AddTransceiverFromKind(
webrtc.RTPCodecTypeVideo,
webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly},
)
require.NoError(t, err)
offer, err := client.CreateOffer(nil)
require.NoError(t, err)
require.NoError(t, client.SetLocalDescription(offer))
offerId++
setCalls := handler.OnSetRemoteDescriptionOfferCallCount()
require.NoError(t, tr.HandleRemoteDescription(offer, offerId))
require.Eventually(t, func() bool {
return handler.OnSetRemoteDescriptionOfferCallCount() > setCalls
}, 5*time.Second, 10*time.Millisecond)
parsed := tr.RemoteOfferParsed()
require.NotNil(t, parsed)
require.Len(t, parsed.MediaDescriptions, numSections)
var want []string
for _, tc := range client.GetTransceivers() {
want = append(want, tc.Mid())
}
require.Equal(t, want, mids(parsed))
// wait for the answer so that the client can offer again
require.Eventually(t, func() bool {
return client.SignalingState() == webrtc.SignalingStateStable
}, 5*time.Second, 10*time.Millisecond)
}
sendOffer(1)
sendOffer(2)
// an offer that does not parse fails the negotiation and leaves the last parse alone
before := tr.RemoteOfferParsed()
failures := handler.OnNegotiationFailedCallCount()
require.NoError(t, tr.HandleRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: "not an sdp"}, offerId+1))
require.Eventually(t, func() bool {
return handler.OnNegotiationFailedCallCount() > failures
}, 5*time.Second, 10*time.Millisecond)
require.Same(t, before, tr.RemoteOfferParsed())
}
// in one-shot signalling mode an ICE restart arrives as an SDP fragment that is patched into
// the remote offer and set on pion directly, the stored parse must follow it
func TestRemoteOfferParsedFollowsICERestartFragment(t *testing.T) {
codecs := []*livekit.Codec{{Mime: mime.MimeTypeVP8.String()}}
handler := &transportfakes.FakeHandler{}
tr, err := NewPCTransport(TransportParams{
Config: &WebRTCConfig{
BufferFactory: buffer.NewFactoryOfBufferFactory(500, 200).CreateBufferFactory(),
},
EnabledPublishCodecs: codecs,
IsOfferer: false,
UseOneShotSignallingMode: true,
Handler: handler,
})
require.NoError(t, err)
defer tr.Close()
var clientME webrtc.MediaEngine
require.NoError(t, registerCodecs(&clientME, codecs, RTCPFeedbackConfig{}, false))
client, err := webrtc.NewAPI(webrtc.WithMediaEngine(&clientME)).NewPeerConnection(webrtc.Configuration{})
require.NoError(t, err)
defer client.Close()
_, err = client.AddTransceiverFromKind(
webrtc.RTPCodecTypeVideo,
webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly},
)
require.NoError(t, err)
offer, err := client.CreateOffer(nil)
require.NoError(t, err)
require.NoError(t, client.SetLocalDescription(offer))
require.NoError(t, tr.HandleRemoteDescription(offer, 1))
before := tr.RemoteOfferParsed()
require.NotNil(t, before)
ufrag, _, err := lksdp.ExtractICECredential(before)
require.NoError(t, err)
require.NotEqual(t, "restartUfrag", ufrag)
_, _, err = tr.GetAnswer()
require.NoError(t, err)
mid := client.GetTransceivers()[0].Mid()
fragment := "m=video 9 UDP/TLS/RTP/SAVPF 96\r\n" +
"a=mid:" + mid + "\r\n" +
"a=ice-ufrag:restartUfrag\r\n" +
"a=ice-pwd:restartPwd0123456789abcdef\r\n" +
"a=candidate:1 1 udp 2122260223 192.0.2.1 61764 typ host generation 0\r\n"
answer, err := tr.HandleICERestartSDPFragment(fragment)
require.NoError(t, err)
require.NotEmpty(t, answer)
after := tr.RemoteOfferParsed()
require.NotSame(t, before, after)
ufrag, pwd, err := lksdp.ExtractICECredential(after)
require.NoError(t, err)
require.Equal(t, "restartUfrag", ufrag)
require.Equal(t, "restartPwd0123456789abcdef", pwd)
}
+3 -6
View File
@@ -523,12 +523,9 @@ func (t *TransportManager) GetUnmatchMediaForOffer(parsedOffer *sdp.SessionDescr
return
}
func (t *TransportManager) LastPublisherOffer() *webrtc.SessionDescription {
return t.publisher.CurrentRemoteDescription()
}
func (t *TransportManager) LastPublisherOfferPending() *webrtc.SessionDescription {
return t.publisher.PendingRemoteDescription()
// LastPublisherOfferParsed returns the parsed form of the last offer set on the publisher peer connection
func (t *TransportManager) LastPublisherOfferParsed() *sdp.SessionDescription {
return t.publisher.RemoteOfferParsed()
}
func (t *TransportManager) HandleOffer(offer webrtc.SessionDescription, offerId uint32, shouldPend bool) error {