Drop remote candidates based on lite option (#3127)

Only drop remote candidates if remote peer is
not lite and local peer is lite.

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

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 {