fixed potentially skipping offer in PCTransport

this could explain the "missing" track problem when the server fails to send some tracks to some of the subscribers

pkg/rtc/config.go

@@ -152,7 +152,7 @@ func checkUDPReadBuffer() (int, error) {
 		return 0, err
 	}
 	defer func() { _ = conn.Close() }()
-	_ = conn.SetReadBuffer(minUDPBufferSize)
+	_ = conn.SetReadBuffer(defaultUDPBufferSize)
 	fd, err := conn.File()
 	if err != nil {
 		return 0, nil

pkg/rtc/transport.go

@@ -144,8 +144,10 @@ func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error {
 	state := t.negotiationState
 	t.negotiationState = negotiationStateNone
 	if state == negotiationRetry {
-		// need to Negotiate again
-		t.Negotiate()
+		// need to Negotiate again, do it immediately
+		if err := t.CreateAndSendOffer(nil); err != nil {
+			logger.Errorw("could not negotiate", err)
+		}
 	}
 
 	return nil
@@ -199,6 +201,9 @@ func (t *PCTransport) CreateAndSendOffer(options *webrtc.OfferOptions) error {
 			t.negotiationState = negotiationRetry
 			return nil
 		}
+	} else if t.negotiationState == negotiationRetry {
+		// already set to retry, we can safely skip this attempt
+		return nil
 	}
 
 	offer, err := t.pc.CreateOffer(options)

pkg/rtc/transport_test.go

@@ -1,6 +1,7 @@
 package rtc
 
 import (
+	"sync/atomic"
 	"testing"
 
 	"github.com/livekit/livekit-server/pkg/testutils"
@@ -23,20 +24,8 @@ func TestMissingAnswerDuringICERestart(t *testing.T) {
 	require.NoError(t, err)
 
 	// exchange ICE
-	transportA.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
-		if candidate == nil {
-			return
-		}
-		t.Logf("got ICE candidate from A: %v", candidate)
-		require.NoError(t, transportB.AddICECandidate(candidate.ToJSON()))
-	})
-	transportB.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
-		if candidate == nil {
-			return
-		}
-		t.Logf("got ICE candidate from B: %v", candidate)
-		require.NoError(t, transportA.AddICECandidate(candidate.ToJSON()))
-	})
+	handleICEExchange(t, transportA, transportB)
+
 	// set offer/answer
 	handleOffer := handleOfferFunc(t, transportA, transportB)
 	transportA.OnOffer(handleOffer)
@@ -71,6 +60,38 @@ func TestMissingAnswerDuringICERestart(t *testing.T) {
 	})
 }
 
+func TestNegotiationTiming(t *testing.T) {
+	params := TransportParams{
+		Target: livekit.SignalTarget_SUBSCRIBER,
+		Config: &WebRTCConfig{},
+		Stats:  nil,
+	}
+	transportA, err := NewPCTransport(params)
+	require.NoError(t, err)
+	_, err = transportA.pc.CreateDataChannel("test", nil)
+	require.NoError(t, err)
+	transportB, err := NewPCTransport(params)
+	require.NoError(t, err)
+
+	handleICEExchange(t, transportA, transportB)
+	offer := atomic.Value{}
+	transportA.OnOffer(func(sd webrtc.SessionDescription) {
+		offer.Store(sd)
+	})
+
+	// initial offer
+	require.NoError(t, transportA.CreateAndSendOffer(nil))
+	require.Equal(t, negotiationStateClient, transportA.negotiationState)
+
+	// second try, should've flipped transport status to retry
+	require.NoError(t, transportA.CreateAndSendOffer(nil))
+	require.Equal(t, negotiationRetry, transportA.negotiationState)
+
+	// third try, should've stayed at retry
+	require.NoError(t, transportA.CreateAndSendOffer(nil))
+	require.Equal(t, negotiationRetry, transportA.negotiationState)
+}
+
 func handleOfferFunc(t *testing.T, current, other *PCTransport) func(sd webrtc.SessionDescription) {
 	return func(sd webrtc.SessionDescription) {
 		t.Logf("handling offer")
@@ -84,3 +105,20 @@ func handleOfferFunc(t *testing.T, current, other *PCTransport) func(sd webrtc.S
 		require.NoError(t, current.SetRemoteDescription(answer))
 	}
 }
+
+func handleICEExchange(t *testing.T, a, b *PCTransport) {
+	a.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
+		if candidate == nil {
+			return
+		}
+		t.Logf("got ICE candidate from A: %v", candidate)
+		require.NoError(t, b.AddICECandidate(candidate.ToJSON()))
+	})
+	b.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
+		if candidate == nil {
+			return
+		}
+		t.Logf("got ICE candidate from B: %v", candidate)
+		require.NoError(t, a.AddICECandidate(candidate.ToJSON()))
+	})
+}