diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 71c8949cc..798fd8cbd 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -178,6 +178,8 @@ type PCTransport struct { firstOfferReceived bool firstOfferNoDataChannel bool + remoteICEIsLite *bool + localICEIsLite *bool reliableDC *webrtc.DataChannel reliableDCOpened bool lossyDC *webrtc.DataChannel @@ -1401,7 +1403,7 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { c := e.data.(*webrtc.ICECandidateInit) filtered := false - if t.params.DropRemoteICECandidates || (t.preferTCP.Load() && !strings.Contains(c.Candidate, "tcp")) { + if t.preferTCP.Load() && !strings.Contains(c.Candidate, "tcp") { t.params.Logger.Debugw("filtering out remote candidate", "candidate", c.Candidate) filtered = true } @@ -1421,6 +1423,11 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { return nil } + if t.params.DropRemoteICECandidates { + t.params.Logger.Debugw("dropping remote ICE candidate", "candidate", c.Candidate) + return nil + } + if err := t.pc.AddICECandidate(*c); err != nil { t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c) return errors.Wrap(err, "add ice candidate failed") @@ -1445,6 +1452,25 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, return sd } + _, iceLite := parsed.Attribute("ice-lite") + var liteSet bool + if isLocal { + if t.localICEIsLite == nil { + t.localICEIsLite = &iceLite + liteSet = true + } + } else { + if t.remoteICEIsLite == nil { + t.remoteICEIsLite = &iceLite + liteSet = true + } + } + if liteSet && t.localICEIsLite != nil && t.remoteICEIsLite != nil { + // only drop remote candidates if local is lite and remote is not + t.params.DropRemoteICECandidates = t.params.DropRemoteICECandidates && (*t.localICEIsLite && !*t.remoteICEIsLite) + t.params.Logger.Debugw("setting DropRemoteICECandidates", "dropRemoteCandidate", t.params.DropRemoteICECandidates, "localICELite", *t.localICEIsLite, "remoteICELite", *t.remoteICEIsLite) + } + filterAttributes := func(attrs []sdp.Attribute) []sdp.Attribute { filteredAttrs := make([]sdp.Attribute, 0, len(attrs)) for _, a := range attrs { @@ -1673,6 +1699,9 @@ func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error { } for _, c := range t.pendingRemoteCandidates { + if t.params.DropRemoteICECandidates { + continue + } if err := t.pc.AddICECandidate(*c); err != nil { t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c) return errors.Wrap(err, "add ice candidate failed") diff --git a/pkg/rtc/transport_test.go b/pkg/rtc/transport_test.go index c80b206ef..e475d32a3 100644 --- a/pkg/rtc/transport_test.go +++ b/pkg/rtc/transport_test.go @@ -28,6 +28,7 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/transport" "github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes" + "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/testutils" "github.com/livekit/protocol/livekit" ) @@ -503,6 +504,115 @@ func TestFilteringCandidates(t *testing.T) { transport.Close() } +func TestDropRemoteICECandidates(t *testing.T) { + cases := []struct { + name string + remoteLite bool + localLite bool + expecteLocalDrop bool + expecteRemoteDrop bool + }{ + { + name: "both not lite", + localLite: false, + remoteLite: false, + expecteLocalDrop: false, + expecteRemoteDrop: false, + }, + { + name: "remote lite", + localLite: false, + remoteLite: true, + expecteLocalDrop: false, + expecteRemoteDrop: true, + }, + { + name: "local lite", + localLite: true, + remoteLite: false, + expecteLocalDrop: true, + expecteRemoteDrop: false, + }, + { + name: "both lite", + localLite: true, + remoteLite: true, + expecteLocalDrop: false, + expecteRemoteDrop: false, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + params := TransportParams{ + ParticipantID: "id", + ParticipantIdentity: "identity", + Config: &WebRTCConfig{}, + IsOfferer: true, + ProtocolVersion: types.CurrentProtocol, + DropRemoteICECandidates: true, + } + + paramsA := params + paramsA.Config.SettingEngine.SetLite(c.localLite) + handlerA := &transportfakes.FakeHandler{} + paramsA.Handler = handlerA + transportLocal, err := NewPCTransport(paramsA) + require.NoError(t, err) + _, err = transportLocal.pc.CreateDataChannel(LossyDataChannel, nil) + require.NoError(t, err) + + paramsB := params + paramsB.Config.SettingEngine.SetLite(c.remoteLite) + handlerB := &transportfakes.FakeHandler{} + paramsB.Handler = handlerB + paramsB.IsOfferer = false + transportRemote, err := NewPCTransport(paramsB) + require.NoError(t, err) + + require.False(t, transportLocal.IsEstablished()) + require.False(t, transportRemote.IsEstablished()) + + handleICEExchange(t, transportLocal, transportRemote, handlerA, handlerB) + var offer atomic.Pointer[webrtc.SessionDescription] + handlerA.OnOfferCalls(func(sd webrtc.SessionDescription) error { + parsed, err := sd.Unmarshal() + require.NoError(t, err) + _, lite := parsed.Attribute("ice-lite") + require.Equal(t, c.localLite, lite) + offer.Store(&sd) + return nil + }) + transportLocal.Negotiate(true) + require.Eventually(t, func() bool { + return offer.Load() != nil + }, 100*time.Millisecond, time.Millisecond*10, "offer not received") + + handlerB.OnAnswerCalls(func(sd webrtc.SessionDescription) error { + parsed, err := sd.Unmarshal() + require.NoError(t, err) + _, lite := parsed.Attribute("ice-lite") + require.Equal(t, c.remoteLite, lite, sd.SDP) + transportLocal.HandleRemoteDescription(sd) + return nil + }) + transportRemote.HandleRemoteDescription(*offer.Load()) + + require.Eventually(t, func() bool { + return transportLocal.IsEstablished() + }, 10*time.Second, time.Millisecond*10, "transportA is not established") + require.Eventually(t, func() bool { + return transportRemote.IsEstablished() + }, 10*time.Second, time.Millisecond*10, "transportB is not established") + + require.Equal(t, c.expecteLocalDrop, transportLocal.params.DropRemoteICECandidates) + require.Equal(t, c.expecteRemoteDrop, transportRemote.params.DropRemoteICECandidates) + + transportLocal.Close() + transportRemote.Close() + }) + } +} func handleICEExchange(t *testing.T, a, b *PCTransport, ah, bh *transportfakes.FakeHandler) { ah.OnICECandidateCalls(func(candidate *webrtc.ICECandidate, target livekit.SignalTarget) error {