Merge remote-tracking branch 'origin/master' into raja_min_packets

go.mod

@@ -19,7 +19,7 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
 	github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598
-	github.com/livekit/protocol v1.26.1-0.20241022031344-538889e5de0a
+	github.com/livekit/protocol v1.27.1-0.20241022061022-caa595ed3292
 	github.com/livekit/psrpc v0.6.1-0.20240924010758-9f0a4268a3b9
 	github.com/mackerelio/go-osstat v0.2.5
 	github.com/magefile/mage v1.15.0
@@ -56,6 +56,8 @@ require (
 	gopkg.in/yaml.v3 v3.0.1
 )
 
+// replace github.com/livekit/protocol => ../protocol
+
 require (
 	buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.33.0-20240401165935-b983156c5e99.1 // indirect
 	dario.cat/mergo v1.0.0 // indirect

go.sum

@@ -165,8 +165,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD
 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598 h1:yLlkHk2feSLHstD9n4VKg7YEBR4rLODTI4WE8gNBEnQ=
 github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598/go.mod h1:jwKUCmObuiEDH0iiuJHaGMXwRs3RjrB4G6qqgkr/5oE=
-github.com/livekit/protocol v1.26.1-0.20241022031344-538889e5de0a h1:31YXXJLEwCflp7KEe9rRAwmONyCwHFujTl4MdxegTxw=
-github.com/livekit/protocol v1.26.1-0.20241022031344-538889e5de0a/go.mod h1:nxRzmQBKSYK64gqr7ABWwt78hvrgiO2wYuCojRYb7Gs=
+github.com/livekit/protocol v1.27.1-0.20241022061022-caa595ed3292 h1:wVzOLGSjJpCsdKHKKpPxYhXW/JL90l0XYFQbeINSdP4=
+github.com/livekit/protocol v1.27.1-0.20241022061022-caa595ed3292/go.mod h1:nxRzmQBKSYK64gqr7ABWwt78hvrgiO2wYuCojRYb7Gs=
 github.com/livekit/psrpc v0.6.1-0.20240924010758-9f0a4268a3b9 h1:33oBjGpVD9tYkDXQU42tnHl8eCX9G6PVUToBVuCUyOs=
 github.com/livekit/psrpc v0.6.1-0.20240924010758-9f0a4268a3b9/go.mod h1:CQUBSPfYYAaevg1TNCc6/aYsa8DJH4jSRFdCeSZk5u0=
 github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o=

pkg/agent/testutils/server.go

@@ -18,7 +18,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/service"
 	"github.com/livekit/protocol/auth"
 	"github.com/livekit/protocol/livekit"
-	"github.com/livekit/protocol/utils"
+	"github.com/livekit/protocol/utils/events"
 	"github.com/livekit/protocol/utils/guid"
 	"github.com/livekit/protocol/utils/must"
 	"github.com/livekit/protocol/utils/options"
@@ -111,11 +111,11 @@ func (h *TestServer) SimulateAgentWorker(opts ...SimulatedWorkerOption) *AgentWo
 		jobs:                   map[string]*AgentJob{},
 		SimulatedWorkerOptions: o,
 
-		RegisterWorkerResponses: utils.NewDefaultEventObserverList[*livekit.RegisterWorkerResponse](),
-		AvailabilityRequests:    utils.NewDefaultEventObserverList[*livekit.AvailabilityRequest](),
-		JobAssignments:          utils.NewDefaultEventObserverList[*livekit.JobAssignment](),
-		JobTerminations:         utils.NewDefaultEventObserverList[*livekit.JobTermination](),
-		WorkerPongs:             utils.NewDefaultEventObserverList[*livekit.WorkerPong](),
+		RegisterWorkerResponses: events.NewObserverList[*livekit.RegisterWorkerResponse](),
+		AvailabilityRequests:    events.NewObserverList[*livekit.AvailabilityRequest](),
+		JobAssignments:          events.NewObserverList[*livekit.JobAssignment](),
+		JobTerminations:         events.NewObserverList[*livekit.JobTermination](),
+		WorkerPongs:             events.NewObserverList[*livekit.WorkerPong](),
 	}
 	w.ctx, w.cancel = context.WithCancel(context.Background())
 
@@ -178,11 +178,11 @@ type AgentWorker struct {
 	serverMessages deque.Deque[*livekit.ServerMessage]
 	jobs           map[string]*AgentJob
 
-	RegisterWorkerResponses *utils.EventObserverList[*livekit.RegisterWorkerResponse]
-	AvailabilityRequests    *utils.EventObserverList[*livekit.AvailabilityRequest]
-	JobAssignments          *utils.EventObserverList[*livekit.JobAssignment]
-	JobTerminations         *utils.EventObserverList[*livekit.JobTermination]
-	WorkerPongs             *utils.EventObserverList[*livekit.WorkerPong]
+	RegisterWorkerResponses *events.ObserverList[*livekit.RegisterWorkerResponse]
+	AvailabilityRequests    *events.ObserverList[*livekit.AvailabilityRequest]
+	JobAssignments          *events.ObserverList[*livekit.JobAssignment]
+	JobTerminations         *events.ObserverList[*livekit.JobTermination]
+	WorkerPongs             *events.ObserverList[*livekit.WorkerPong]
 }
 
 func (w *AgentWorker) statusWorker() {

pkg/config/config.go

@@ -253,8 +253,8 @@ type RoomConfig struct {
 	// deprecated, moved to limits
 	MaxRoomNameLength int `yaml:"max_room_name_length,omitempty"`
 	// deprecated, moved to limits
-	MaxParticipantIdentityLength int                                  `yaml:"max_participant_identity_length,omitempty"`
-	RoomConfigurations           map[string]livekit.RoomConfiguration `yaml:"room_configurations,omitempty"`
+	MaxParticipantIdentityLength int                                   `yaml:"max_participant_identity_length,omitempty"`
+	RoomConfigurations           map[string]*livekit.RoomConfiguration `yaml:"room_configurations,omitempty"`
 }
 
 type CodecSpec struct {

pkg/routing/interfaces.go

@@ -169,6 +169,7 @@ func ParticipantInitFromStartSession(ss *livekit.StartSession, region string) (*
 		AdaptiveStream:  ss.AdaptiveStream,
 		ID:              livekit.ParticipantID(ss.ParticipantId),
 		DisableICELite:  ss.DisableIceLite,
+		CreateRoom:      ss.CreateRoom,
 	}
 	if ss.SubscriberAllowPause != nil {
 		subscriberAllowPause := *ss.SubscriberAllowPause

pkg/rtc/participant.go

@@ -1334,7 +1334,7 @@ type AnyTransportHandler struct {
 	p *ParticipantImpl
 }
 
-func (h AnyTransportHandler) OnFailed(isShortLived bool) {
+func (h AnyTransportHandler) OnFailed(_isShortLived bool, _ici *types.ICEConnectionInfo) {
 	h.p.onAnyTransportFailed()
 }
 

pkg/rtc/room.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"net"
 	"slices"
 	"sort"
 	"strings"
@@ -54,6 +55,8 @@ const (
 	dataForwardLoadBalanceThreshold = 20
 
 	simulateDisconnectSignalTimeout = 5 * time.Second
+
+	minIPTruncateLen = 8
 )
 
 var (
@@ -1728,9 +1731,7 @@ func (r *Room) createAgentDispatchesFromRoomAgent() {
 	roomDisp := r.internal.AgentDispatches
 	if len(roomDisp) == 0 {
 		// Backward compatibility: by default, start any agent in the empty JobName
-		roomDisp = []*livekit.RoomAgentDispatch{
-			&livekit.RoomAgentDispatch{},
-		}
+		roomDisp = []*livekit.RoomAgentDispatch{{}}
 	}
 
 	for _, ag := range roomDisp {
@@ -1818,8 +1819,8 @@ func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} {
 		candidates := make([]string, 0, len(info.Remote)+len(info.Local))
 		for _, c := range info.Local {
 			cStr := "[local]"
-			if c.Selected {
-				cStr += "[selected]"
+			if c.SelectedOrder != 0 {
+				cStr += fmt.Sprintf("[selected:%d]", c.SelectedOrder)
 			} else if c.Filtered {
 				cStr += "[filtered]"
 			}
@@ -1831,15 +1832,35 @@ func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} {
 		}
 		for _, c := range info.Remote {
 			cStr := "[remote]"
-			if c.Selected {
-				cStr += "[selected]"
+			if c.SelectedOrder != 0 {
+				cStr += fmt.Sprintf("[selected:%d]", c.SelectedOrder)
 			} else if c.Filtered {
 				cStr += "[filtered]"
 			}
 			if c.Trickle {
 				cStr += "[trickle]"
 			}
-			cStr += " " + c.Remote.String()
+			remoteAddress := c.Remote.Address()
+			ipAddr := net.ParseIP(remoteAddress)
+			isPrivate := false
+			if ipAddr != nil {
+				isPrivate = ipAddr.IsPrivate()
+			}
+			if !isPrivate && len(remoteAddress) > minIPTruncateLen {
+				remoteAddress = remoteAddress[:len(remoteAddress)-3] + "..."
+			}
+			cStr += " " + fmt.Sprintf("%s %s %s:%d", c.Remote.NetworkType(), c.Remote.Type(), remoteAddress, c.Remote.Port())
+			if relatedAddress := c.Remote.RelatedAddress(); relatedAddress != nil {
+				ipAddr = net.ParseIP(relatedAddress.Address)
+				if ipAddr != nil {
+					isPrivate = ipAddr.IsPrivate()
+					relatedAddressAddress := relatedAddress.Address
+					if !isPrivate && len(relatedAddressAddress) > minIPTruncateLen {
+						relatedAddressAddress = relatedAddressAddress[:len(relatedAddressAddress)-3] + "..."
+					}
+					cStr += " " + fmt.Sprintf(" related %s:%d", relatedAddressAddress, relatedAddress.Port)
+				}
+			}
 			candidates = append(candidates, cStr)
 		}
 		if len(candidates) > 0 {

pkg/rtc/transport.go

@@ -152,6 +152,10 @@ func (w wrappedICECandidatePairLogger) MarshalLogObject(e zapcore.ObjectEncoder)
 		e.AddString("remoteCandidateType", w.pair.Remote.Typ.String())
 		e.AddString("remoteAdddress", w.pair.Remote.Address[:len(w.pair.Remote.Address)-3]+"...")
 		e.AddUint16("remotePort", w.pair.Remote.Port)
+		if w.pair.Remote.RelatedAddress != "" {
+			e.AddString("relatedAdddress", w.pair.Remote.RelatedAddress[:len(w.pair.Remote.RelatedAddress)-3]+"...")
+			e.AddUint16("relatedPort", w.pair.Remote.RelatedPort)
+		}
 	}
 	return nil
 }
@@ -178,6 +182,8 @@ type PCTransport struct {
 
 	firstOfferReceived      bool
 	firstOfferNoDataChannel bool
+	remoteICEIsLite         *bool
+	localICEIsLite          *bool
 	reliableDC              *webrtc.DataChannel
 	reliableDCOpened        bool
 	lossyDC                 *webrtc.DataChannel
@@ -291,7 +297,11 @@ func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimat
 	//
 	se.DisableSRTPReplayProtection(true)
 	se.DisableSRTCPReplayProtection(true)
-	if !params.ProtocolVersion.SupportsICELite() {
+	if !params.ProtocolVersion.SupportsICELite() || !params.ClientInfo.SupportPrflxOverRelay() {
+		// if client don't support prflx over relay which is only Firefox, disable ICE Lite to ensure that
+		// dropping remote ICE candidates does not get enabled. Firefox does aggressive nomination and
+		// dropping remote ICE candidates means server would accept all switches and it could end up with
+		// the lower priority candidate. As Firefox does not support migration, ICE Lite can be disabled.
 		se.SetLite(false)
 	}
 	se.SetDTLSRetransmissionInterval(dtlsRetransmissionInterval)
@@ -471,8 +481,12 @@ func (t *PCTransport) createPeerConnection() error {
 	t.pc.SCTP().Transport().ICETransport().OnSelectedCandidatePairChange(func(pair *webrtc.ICECandidatePair) {
 		t.params.Logger.Debugw("selected ICE candidate pair changed", "pair", wrappedICECandidatePairLogger{pair})
 		t.connectionDetails.SetSelectedPair(pair)
-		if t.selectedPair.Load() != nil {
-			t.params.Logger.Infow("ice reconnected or switched pair", "pair", wrappedICECandidatePairLogger{pair})
+		existingPair := t.selectedPair.Load()
+		if existingPair != nil {
+			t.params.Logger.Infow(
+				"ice reconnected or switched pair",
+				"existingPair", wrappedICECandidatePairLogger{existingPair},
+				"newPair", wrappedICECandidatePairLogger{pair})
 		}
 		t.selectedPair.Store(pair)
 	})
@@ -630,7 +644,7 @@ func (t *PCTransport) handleConnectionFailed(forceShortConn bool) {
 		}
 	}
 
-	t.params.Handler.OnFailed(isShort)
+	t.params.Handler.OnFailed(isShort, t.GetICEConnectionInfo())
 }
 
 func (t *PCTransport) onICEConnectionStateChange(state webrtc.ICEConnectionState) {
@@ -1401,7 +1415,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
 	}
@@ -1411,7 +1425,7 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error {
 		filtered = true
 	}
 
-	t.connectionDetails.AddRemoteCandidate(*c, filtered, true)
+	t.connectionDetails.AddRemoteCandidate(*c, filtered, true, false)
 	if filtered {
 		return nil
 	}
@@ -1421,6 +1435,12 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error {
 		return nil
 	}
 
+	if t.params.DropRemoteICECandidates {
+		t.params.Logger.Debugw("dropping remote ICE candidate", "candidate", c.Candidate)
+		t.connectionDetails.AddRemoteCandidate(*c, true, true, true)
+		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 +1465,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 {
@@ -1468,7 +1507,7 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP,
 				if isLocal {
 					t.connectionDetails.AddLocalICECandidate(c, excluded, false)
 				} else {
-					t.connectionDetails.AddRemoteICECandidate(c, excluded, false)
+					t.connectionDetails.AddRemoteICECandidate(c, excluded, false, false)
 				}
 			} else {
 				filteredAttrs = append(filteredAttrs, a)
@@ -1673,6 +1712,10 @@ func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error {
 	}
 
 	for _, c := range t.pendingRemoteCandidates {
+		if t.params.DropRemoteICECandidates {
+			t.connectionDetails.AddRemoteCandidate(*c, true, true, true)
+			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/handler.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/pion/webrtc/v3"
 
+	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
 	"github.com/livekit/protocol/livekit"
 )
@@ -36,7 +37,7 @@ type Handler interface {
 	OnICECandidate(c *webrtc.ICECandidate, target livekit.SignalTarget) error
 	OnInitialConnected()
 	OnFullyEstablished()
-	OnFailed(isShortLived bool)
+	OnFailed(isShortLived bool, iceConnectionInfo *types.ICEConnectionInfo)
 	OnTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver)
 	OnDataPacket(kind livekit.DataPacket_Kind, data []byte)
 	OnDataSendError(err error)

pkg/rtc/transport/transportfakes/fake_handler.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/livekit/livekit-server/pkg/rtc/transport"
+	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
 	"github.com/livekit/protocol/livekit"
 	webrtc "github.com/pion/webrtc/v3"
@@ -33,10 +34,11 @@ type FakeHandler struct {
 	onDataSendErrorArgsForCall []struct {
 		arg1 error
 	}
-	OnFailedStub        func(bool)
+	OnFailedStub        func(bool, *types.ICEConnectionInfo)
 	onFailedMutex       sync.RWMutex
 	onFailedArgsForCall []struct {
 		arg1 bool
+		arg2 *types.ICEConnectionInfo
 	}
 	OnFullyEstablishedStub        func()
 	onFullyEstablishedMutex       sync.RWMutex
@@ -230,16 +232,17 @@ func (fake *FakeHandler) OnDataSendErrorArgsForCall(i int) error {
 	return argsForCall.arg1
 }
 
-func (fake *FakeHandler) OnFailed(arg1 bool) {
+func (fake *FakeHandler) OnFailed(arg1 bool, arg2 *types.ICEConnectionInfo) {
 	fake.onFailedMutex.Lock()
 	fake.onFailedArgsForCall = append(fake.onFailedArgsForCall, struct {
 		arg1 bool
-	}{arg1})
+		arg2 *types.ICEConnectionInfo
+	}{arg1, arg2})
 	stub := fake.OnFailedStub
-	fake.recordInvocation("OnFailed", []interface{}{arg1})
+	fake.recordInvocation("OnFailed", []interface{}{arg1, arg2})
 	fake.onFailedMutex.Unlock()
 	if stub != nil {
-		fake.OnFailedStub(arg1)
+		fake.OnFailedStub(arg1, arg2)
 	}
 }
 
@@ -249,17 +252,17 @@ func (fake *FakeHandler) OnFailedCallCount() int {
 	return len(fake.onFailedArgsForCall)
 }
 
-func (fake *FakeHandler) OnFailedCalls(stub func(bool)) {
+func (fake *FakeHandler) OnFailedCalls(stub func(bool, *types.ICEConnectionInfo)) {
 	fake.onFailedMutex.Lock()
 	defer fake.onFailedMutex.Unlock()
 	fake.OnFailedStub = stub
 }
 
-func (fake *FakeHandler) OnFailedArgsForCall(i int) bool {
+func (fake *FakeHandler) OnFailedArgsForCall(i int) (bool, *types.ICEConnectionInfo) {
 	fake.onFailedMutex.RLock()
 	defer fake.onFailedMutex.RUnlock()
 	argsForCall := fake.onFailedArgsForCall[i]
-	return argsForCall.arg1
+	return argsForCall.arg1, argsForCall.arg2
 }
 
 func (fake *FakeHandler) OnFullyEstablished() {

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 {

pkg/rtc/transportmanager.go

@@ -51,16 +51,24 @@ const (
 	udpLossUnstableCountThreshold = 20
 )
 
+// -------------------------------
+
 type TransportManagerTransportHandler struct {
 	transport.Handler
-	t *TransportManager
+	t      *TransportManager
+	logger logger.Logger
 }
 
-func (h TransportManagerTransportHandler) OnFailed(isShortLived bool) {
+func (h TransportManagerTransportHandler) OnFailed(isShortLived bool, iceConnectionInfo *types.ICEConnectionInfo) {
+	if isShortLived {
+		h.logger.Infow("short ice connection", connectionDetailsFields([]*types.ICEConnectionInfo{iceConnectionInfo})...)
+	}
 	h.t.handleConnectionFailed(isShortLived)
-	h.Handler.OnFailed(isShortLived)
+	h.Handler.OnFailed(isShortLived, iceConnectionInfo)
 }
 
+// -------------------------------
+
 type TransportManagerPublisherTransportHandler struct {
 	TransportManagerTransportHandler
 }
@@ -70,6 +78,8 @@ func (h TransportManagerPublisherTransportHandler) OnAnswer(sd webrtc.SessionDes
 	return h.Handler.OnAnswer(sd)
 }
 
+// -------------------------------
+
 type TransportManagerParams struct {
 	Identity                     livekit.ParticipantIdentity
 	SID                          livekit.ParticipantID
@@ -133,6 +143,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
 	}
 	t.mediaLossProxy.OnMediaLossUpdate(t.onMediaLossUpdate)
 
+	lgr := LoggerWithPCTarget(params.Logger, livekit.SignalTarget_PUBLISHER)
 	publisher, err := NewPCTransport(TransportParams{
 		ParticipantID:           params.SID,
 		ParticipantIdentity:     params.Identity,
@@ -142,11 +153,11 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
 		DirectionConfig:         params.Config.Publisher,
 		CongestionControlConfig: params.CongestionControlConfig,
 		EnabledCodecs:           params.EnabledPublishCodecs,
-		Logger:                  LoggerWithPCTarget(params.Logger, livekit.SignalTarget_PUBLISHER),
+		Logger:                  lgr,
 		SimTracks:               params.SimTracks,
 		ClientInfo:              params.ClientInfo,
 		Transport:               livekit.SignalTarget_PUBLISHER,
-		Handler:                 TransportManagerPublisherTransportHandler{TransportManagerTransportHandler{params.PublisherHandler, t}},
+		Handler:                 TransportManagerPublisherTransportHandler{TransportManagerTransportHandler{params.PublisherHandler, t, lgr}},
 		DropRemoteICECandidates: params.DropRemoteICECandidates,
 	})
 	if err != nil {
@@ -154,6 +165,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
 	}
 	t.publisher = publisher
 
+	lgr = LoggerWithPCTarget(params.Logger, livekit.SignalTarget_SUBSCRIBER)
 	subscriber, err := NewPCTransport(TransportParams{
 		ParticipantID:                params.SID,
 		ParticipantIdentity:          params.Identity,
@@ -162,14 +174,14 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
 		DirectionConfig:              params.Config.Subscriber,
 		CongestionControlConfig:      params.CongestionControlConfig,
 		EnabledCodecs:                params.EnabledSubscribeCodecs,
-		Logger:                       LoggerWithPCTarget(params.Logger, livekit.SignalTarget_SUBSCRIBER),
+		Logger:                       lgr,
 		ClientInfo:                   params.ClientInfo,
 		IsOfferer:                    true,
 		IsSendSide:                   true,
 		AllowPlayoutDelay:            params.AllowPlayoutDelay,
 		DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount,
 		Transport:                    livekit.SignalTarget_SUBSCRIBER,
-		Handler:                      TransportManagerTransportHandler{params.SubscriberHandler, t},
+		Handler:                      TransportManagerTransportHandler{params.SubscriberHandler, t, lgr},
 		DropRemoteICECandidates:      params.DropRemoteICECandidates,
 	})
 	if err != nil {
@@ -690,7 +702,7 @@ func (t *TransportManager) onMediaLossUpdate(loss uint8) {
 				t.lock.Unlock()
 
 				t.params.Logger.Infow("udp connection unstable, switch to tcp", "signalingRTT", t.signalingRTT)
-				t.params.SubscriberHandler.OnFailed(true)
+				t.params.SubscriberHandler.OnFailed(true, t.subscriber.GetICEConnectionInfo())
 				return
 			}
 		}

pkg/rtc/types/ice.go

@@ -40,11 +40,11 @@ const (
 
 type ICECandidateExtended struct {
 	// only one of local or remote is set. This is due to type foo in Pion
-	Local    *webrtc.ICECandidate
-	Remote   ice.Candidate
-	Selected bool
-	Filtered bool
-	Trickle  bool
+	Local         *webrtc.ICECandidate
+	Remote        ice.Candidate
+	SelectedOrder int
+	Filtered      bool
+	Trickle       bool
 }
 
 // --------------------------------------------
@@ -64,8 +64,9 @@ func (i *ICEConnectionInfo) HasCandidates() bool {
 
 type ICEConnectionDetails struct {
 	ICEConnectionInfo
-	lock   sync.Mutex
-	logger logger.Logger
+	lock          sync.Mutex
+	selectedCount int
+	logger        logger.Logger
 }
 
 func NewICEConnectionDetails(transport livekit.SignalTarget, l logger.Logger) *ICEConnectionDetails {
@@ -90,18 +91,18 @@ func (d *ICEConnectionDetails) GetInfo() *ICEConnectionInfo {
 	}
 	for _, c := range d.Local {
 		info.Local = append(info.Local, &ICECandidateExtended{
-			Local:    c.Local,
-			Filtered: c.Filtered,
-			Selected: c.Selected,
-			Trickle:  c.Trickle,
+			Local:         c.Local,
+			Filtered:      c.Filtered,
+			SelectedOrder: c.SelectedOrder,
+			Trickle:       c.Trickle,
 		})
 	}
 	for _, c := range d.Remote {
 		info.Remote = append(info.Remote, &ICECandidateExtended{
-			Remote:   c.Remote,
-			Filtered: c.Filtered,
-			Selected: c.Selected,
-			Trickle:  c.Trickle,
+			Remote:        c.Remote,
+			Filtered:      c.Filtered,
+			SelectedOrder: c.SelectedOrder,
+			Trickle:       c.Trickle,
 		})
 	}
 	return info
@@ -133,16 +134,16 @@ func (d *ICEConnectionDetails) AddLocalICECandidate(c ice.Candidate, filtered, t
 	d.AddLocalCandidate(candidate, filtered, trickle)
 }
 
-func (d *ICEConnectionDetails) AddRemoteCandidate(c webrtc.ICECandidateInit, filtered, trickle bool) {
+func (d *ICEConnectionDetails) AddRemoteCandidate(c webrtc.ICECandidateInit, filtered, trickle, canUpdate bool) {
 	candidate, err := unmarshalICECandidate(c)
 	if err != nil {
 		d.logger.Errorw("could not unmarshal candidate", err, "candidate", c)
 		return
 	}
-	d.AddRemoteICECandidate(candidate, filtered, trickle)
+	d.AddRemoteICECandidate(candidate, filtered, trickle, canUpdate)
 }
 
-func (d *ICEConnectionDetails) AddRemoteICECandidate(candidate ice.Candidate, filtered, trickle bool) {
+func (d *ICEConnectionDetails) AddRemoteICECandidate(candidate ice.Candidate, filtered, trickle, canUpdate bool) {
 	if candidate == nil {
 		// end-of-candidates candidate
 		return
@@ -150,10 +151,14 @@ func (d *ICEConnectionDetails) AddRemoteICECandidate(candidate ice.Candidate, fi
 
 	d.lock.Lock()
 	defer d.lock.Unlock()
-	compFn := func(e *ICECandidateExtended) bool {
+	indexFn := func(e *ICECandidateExtended) bool {
 		return isICECandidateEqualTo(e.Remote, candidate)
 	}
-	if slices.ContainsFunc(d.Remote, compFn) {
+	if idx := slices.IndexFunc(d.Remote, indexFn); idx != -1 {
+		if canUpdate {
+			d.Remote[idx].Filtered = filtered
+			d.Remote[idx].Trickle = trickle
+		}
 		return
 	}
 	d.Remote = append(d.Remote, &ICECandidateExtended{
@@ -174,6 +179,9 @@ func (d *ICEConnectionDetails) Clear() {
 func (d *ICEConnectionDetails) SetSelectedPair(pair *webrtc.ICECandidatePair) {
 	d.lock.Lock()
 	defer d.lock.Unlock()
+
+	d.selectedCount++
+
 	remoteIdx := slices.IndexFunc(d.Remote, func(e *ICECandidateExtended) bool {
 		return isICECandidateEqualToCandidate(e.Remote, pair.Remote)
 	})
@@ -195,7 +203,7 @@ func (d *ICEConnectionDetails) SetSelectedPair(pair *webrtc.ICECandidatePair) {
 		remoteIdx = len(d.Remote) - 1
 	}
 	remote := d.Remote[remoteIdx]
-	remote.Selected = true
+	remote.SelectedOrder = d.selectedCount
 
 	localIdx := slices.IndexFunc(d.Local, func(e *ICECandidateExtended) bool {
 		return isCandidateEqualTo(e.Local, pair.Local)
@@ -206,7 +214,7 @@ func (d *ICEConnectionDetails) SetSelectedPair(pair *webrtc.ICECandidatePair) {
 		return
 	}
 	local := d.Local[localIdx]
-	local.Selected = true
+	local.SelectedOrder = d.selectedCount
 
 	d.Type = ICEConnectionTypeUDP
 	if pair.Remote.Protocol == webrtc.ICEProtocolTCP {

pkg/service/auth.go

@@ -179,16 +179,6 @@ func EnsureCreatePermission(ctx context.Context) error {
 	return nil
 }
 
-func GetRoomConfiguration(ctx context.Context) string {
-
-	claims := GetGrants(ctx)
-
-	if claims == nil || claims.Video == nil {
-		return ""
-	}
-	return claims.Video.RoomConfiguration
-}
-
 func EnsureListPermission(ctx context.Context) error {
 	claims := GetGrants(ctx)
 	if claims == nil || claims.Video == nil || !claims.Video.RoomList {

pkg/service/roomallocator.go

@@ -108,8 +108,8 @@ func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.Cre
 			internal.TrackEgress = req.Egress.Tracks
 		}
 	}
-	if req.Agent != nil {
-		internal.AgentDispatches = req.Agent.Dispatches
+	if req.Agents != nil {
+		internal.AgentDispatches = req.Agents
 	}
 	if req.MinPlayoutDelay > 0 || req.MaxPlayoutDelay > 0 {
 		internal.PlayoutDelay = &livekit.PlayoutDelay{
@@ -200,11 +200,11 @@ func applyDefaultRoomConfig(room *livekit.Room, internal *livekit.RoomInternal,
 }
 
 func (r *StandardRoomAllocator) applyNamedRoomConfiguration(req *livekit.CreateRoomRequest) (*livekit.CreateRoomRequest, error) {
-	if req.ConfigName == "" {
+	if req.RoomPreset == "" {
 		return req, nil
 	}
 
-	conf, ok := r.config.Room.RoomConfigurations[req.ConfigName]
+	conf, ok := r.config.Room.RoomConfigurations[req.RoomPreset]
 	if !ok {
 		return req, psrpc.NewErrorf(psrpc.InvalidArgument, "unknown room confguration in create room request")
 	}
@@ -224,8 +224,11 @@ func (r *StandardRoomAllocator) applyNamedRoomConfiguration(req *livekit.CreateR
 	if clone.Egress == nil {
 		clone.Egress = utils.CloneProto(conf.Egress)
 	}
-	if clone.Agent == nil {
-		clone.Agent = utils.CloneProto(conf.Agent)
+	if clone.Agents == nil {
+		clone.Agents = make([]*livekit.RoomAgentDispatch, 0, len(conf.Agents))
+		for _, agent := range conf.Agents {
+			clone.Agents = append(clone.Agents, utils.CloneProto(agent))
+		}
 	}
 	if clone.MinPlayoutDelay == 0 {
 		clone.MinPlayoutDelay = conf.MinPlayoutDelay

pkg/service/roommanager.go

@@ -278,13 +278,13 @@ func (r *RoomManager) CreateRoom(ctx context.Context, req *livekit.CreateRoomReq
 // StartSession starts WebRTC session when a new participant is connected, takes place on RTC node
 func (r *RoomManager) StartSession(
 	ctx context.Context,
-	createRoom *livekit.CreateRoomRequest,
 	pi routing.ParticipantInit,
 	requestSource routing.MessageSource,
 	responseSink routing.MessageSink,
 ) error {
 	sessionStartTime := time.Now()
 
+	createRoom := pi.CreateRoom
 	room, err := r.getOrCreateRoom(ctx, createRoom)
 	if err != nil {
 		return err
@@ -990,7 +990,9 @@ func (r *RoomManager) refreshToken(participant types.LocalParticipant) error {
 		SetValidFor(tokenDefaultTTL).
 		SetMetadata(grants.Metadata).
 		SetAttributes(grants.Attributes).
-		AddGrant(grants.Video)
+		SetVideoGrant(grants.Video).
+		SetRoomConfig(grants.GetRoomConfiguration()).
+		SetRoomPreset(grants.RoomPreset)
 	jwt, err := token.ToJWT()
 	if err == nil {
 		err = participant.SendRefreshToken(jwt)

pkg/service/rtcservice.go

@@ -166,6 +166,12 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic
 		}
 	}
 
+	createRequest := &livekit.CreateRoomRequest{
+		Name:       string(roomName),
+		RoomPreset: claims.RoomPreset,
+	}
+	SetRoomConfiguration(createRequest, claims.GetRoomConfiguration())
+
 	pi = routing.ParticipantInit{
 		Reconnect:       boolValue(reconnectParam),
 		ReconnectReason: livekit.ReconnectReason(reconnectReason),
@@ -175,10 +181,7 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic
 		Client:          s.ParseClientInfo(r),
 		Grants:          claims,
 		Region:          region,
-		CreateRoom: &livekit.CreateRoomRequest{
-			Name:       string(roomName),
-			ConfigName: GetRoomConfiguration(r.Context()),
-		},
+		CreateRoom:      createRequest,
 	}
 	if pi.Reconnect {
 		pi.ID = livekit.ParticipantID(participantID)

pkg/service/servicefakes/fake_session_handler.go

@@ -12,15 +12,14 @@ import (
 )
 
 type FakeSessionHandler struct {
-	HandleSessionStub        func(context.Context, *livekit.CreateRoomRequest, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error
+	HandleSessionStub        func(context.Context, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error
 	handleSessionMutex       sync.RWMutex
 	handleSessionArgsForCall []struct {
 		arg1 context.Context
-		arg2 *livekit.CreateRoomRequest
-		arg3 routing.ParticipantInit
-		arg4 livekit.ConnectionID
-		arg5 routing.MessageSource
-		arg6 routing.MessageSink
+		arg2 routing.ParticipantInit
+		arg3 livekit.ConnectionID
+		arg4 routing.MessageSource
+		arg5 routing.MessageSink
 	}
 	handleSessionReturns struct {
 		result1 error
@@ -43,23 +42,22 @@ type FakeSessionHandler struct {
 	invocationsMutex sync.RWMutex
 }
 
-func (fake *FakeSessionHandler) HandleSession(arg1 context.Context, arg2 *livekit.CreateRoomRequest, arg3 routing.ParticipantInit, arg4 livekit.ConnectionID, arg5 routing.MessageSource, arg6 routing.MessageSink) error {
+func (fake *FakeSessionHandler) HandleSession(arg1 context.Context, arg2 routing.ParticipantInit, arg3 livekit.ConnectionID, arg4 routing.MessageSource, arg5 routing.MessageSink) error {
 	fake.handleSessionMutex.Lock()
 	ret, specificReturn := fake.handleSessionReturnsOnCall[len(fake.handleSessionArgsForCall)]
 	fake.handleSessionArgsForCall = append(fake.handleSessionArgsForCall, struct {
 		arg1 context.Context
-		arg2 *livekit.CreateRoomRequest
-		arg3 routing.ParticipantInit
-		arg4 livekit.ConnectionID
-		arg5 routing.MessageSource
-		arg6 routing.MessageSink
-	}{arg1, arg2, arg3, arg4, arg5, arg6})
+		arg2 routing.ParticipantInit
+		arg3 livekit.ConnectionID
+		arg4 routing.MessageSource
+		arg5 routing.MessageSink
+	}{arg1, arg2, arg3, arg4, arg5})
 	stub := fake.HandleSessionStub
 	fakeReturns := fake.handleSessionReturns
-	fake.recordInvocation("HandleSession", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6})
+	fake.recordInvocation("HandleSession", []interface{}{arg1, arg2, arg3, arg4, arg5})
 	fake.handleSessionMutex.Unlock()
 	if stub != nil {
-		return stub(arg1, arg2, arg3, arg4, arg5, arg6)
+		return stub(arg1, arg2, arg3, arg4, arg5)
 	}
 	if specificReturn {
 		return ret.result1
@@ -73,17 +71,17 @@ func (fake *FakeSessionHandler) HandleSessionCallCount() int {
 	return len(fake.handleSessionArgsForCall)
 }
 
-func (fake *FakeSessionHandler) HandleSessionCalls(stub func(context.Context, *livekit.CreateRoomRequest, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error) {
+func (fake *FakeSessionHandler) HandleSessionCalls(stub func(context.Context, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error) {
 	fake.handleSessionMutex.Lock()
 	defer fake.handleSessionMutex.Unlock()
 	fake.HandleSessionStub = stub
 }
 
-func (fake *FakeSessionHandler) HandleSessionArgsForCall(i int) (context.Context, *livekit.CreateRoomRequest, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) {
+func (fake *FakeSessionHandler) HandleSessionArgsForCall(i int) (context.Context, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) {
 	fake.handleSessionMutex.RLock()
 	defer fake.handleSessionMutex.RUnlock()
 	argsForCall := fake.handleSessionArgsForCall[i]
-	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5
 }
 
 func (fake *FakeSessionHandler) HandleSessionReturns(result1 error) {

pkg/service/signal.go

@@ -39,7 +39,6 @@ type SessionHandler interface {
 
 	HandleSession(
 		ctx context.Context,
-		createRoom *livekit.CreateRoomRequest,
 		pi routing.ParticipantInit,
 		connectionID livekit.ConnectionID,
 		requestSource routing.MessageSource,
@@ -94,7 +93,6 @@ func (s *defaultSessionHandler) Logger(ctx context.Context) logger.Logger {
 
 func (s *defaultSessionHandler) HandleSession(
 	ctx context.Context,
-	createRoom *livekit.CreateRoomRequest,
 	pi routing.ParticipantInit,
 	connectionID livekit.ConnectionID,
 	requestSource routing.MessageSource,
@@ -102,7 +100,7 @@ func (s *defaultSessionHandler) HandleSession(
 ) error {
 	prometheus.IncrementParticipantRtcInit(1)
 
-	rtcNode, err := s.router.GetNodeForRoom(ctx, livekit.RoomName(createRoom.Name))
+	rtcNode, err := s.router.GetNodeForRoom(ctx, livekit.RoomName(pi.CreateRoom.Name))
 	if err != nil {
 		return err
 	}
@@ -115,7 +113,7 @@ func (s *defaultSessionHandler) HandleSession(
 		return err
 	}
 
-	return s.roomManager.StartSession(ctx, createRoom, pi, requestSource, responseSink)
+	return s.roomManager.StartSession(ctx, pi, requestSource, responseSink)
 }
 
 func (s *SignalServer) Start() error {
@@ -181,19 +179,7 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe
 	// and the delivery of any parting messages from the client. take care to
 	// copy the incoming rpc headers to avoid dropping any session vars.
 	ctx := metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context()))
-
-	createRoom := ss.CreateRoom
-	if createRoom == nil {
-		createRoom = &livekit.CreateRoomRequest{
-			Name: ss.RoomName,
-		}
-
-		if pi.Grants != nil && pi.Grants.Video != nil {
-			createRoom.ConfigName = pi.Grants.Video.RoomConfiguration
-		}
-	}
-
-	err = r.sessionHandler.HandleSession(ctx, createRoom, *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink)
+	err = r.sessionHandler.HandleSession(ctx, *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink)
 	if err != nil {
 		sink.Close()
 		l.Errorw("could not handle new participant", err)

pkg/service/signal_test.go

@@ -67,7 +67,6 @@ func TestSignal(t *testing.T) {
 			LoggerStub: func(context.Context) logger.Logger { return logger.GetLogger() },
 			HandleSessionStub: func(
 				ctx context.Context,
-				createRoom *livekit.CreateRoomRequest,
 				pi routing.ParticipantInit,
 				connectionID livekit.ConnectionID,
 				requestSource routing.MessageSource,
@@ -124,7 +123,6 @@ func TestSignal(t *testing.T) {
 			LoggerStub: func(context.Context) logger.Logger { return logger.GetLogger() },
 			HandleSessionStub: func(
 				ctx context.Context,
-				createRoom *livekit.CreateRoomRequest,
 				pi routing.ParticipantInit,
 				connectionID livekit.ConnectionID,
 				requestSource routing.MessageSource,

pkg/service/sip.go

@@ -392,27 +392,26 @@ func (s *SIPService) CreateSIPParticipantRequest(ctx context.Context, req *livek
 		return nil, ErrSIPNotConnected
 	}
 	callID := sip.NewCallID()
-	log := logger.GetLogger()
-	if projectID != "" {
-		log = log.WithValues("projectID", projectID)
-	}
-	unlikelyLogger := log.WithUnlikelyValues(
+	log := logger.GetLogger().WithUnlikelyValues(
 		"callID", callID,
 		"room", req.RoomName,
 		"sipTrunk", req.SipTrunkId,
 		"toUser", req.SipCallTo,
 	)
+	if projectID != "" {
+		log = log.WithValues("projectID", projectID)
+	}
 
 	trunk, err := s.store.LoadSIPOutboundTrunk(ctx, req.SipTrunkId)
 	if err != nil {
-		unlikelyLogger.Errorw("cannot get trunk to update sip participant", err)
+		log.Errorw("cannot get trunk to update sip participant", err)
 		return nil, err
 	}
 	return rpc.NewCreateSIPParticipantRequest(projectID, callID, host, wsUrl, token, req, trunk)
 }
 
 func (s *SIPService) TransferSIPParticipant(ctx context.Context, req *livekit.TransferSIPParticipantRequest) (*emptypb.Empty, error) {
-	log := logger.GetLogger().WithValues("room", req.RoomName, "participant", req.ParticipantIdentity)
+	log := logger.GetLogger().WithUnlikelyValues("room", req.RoomName, "participant", req.ParticipantIdentity)
 	ireq, err := s.transferSIPParticipantRequest(ctx, req)
 	if err != nil {
 		log.Errorw("cannot create transfer sip participant request", err)

pkg/service/utils.go

@@ -22,6 +22,7 @@ import (
 	"regexp"
 	"strings"
 
+	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 )
 
@@ -67,3 +68,17 @@ func GetClientIP(r *http.Request) string {
 	ip, _, _ := net.SplitHostPort(r.RemoteAddr)
 	return ip
 }
+
+func SetRoomConfiguration(createRequest *livekit.CreateRoomRequest, conf *livekit.RoomConfiguration) {
+	if conf == nil {
+		return
+	}
+	createRequest.Agents = conf.Agents
+	createRequest.Egress = conf.Egress
+	createRequest.EmptyTimeout = conf.EmptyTimeout
+	createRequest.DepartureTimeout = conf.DepartureTimeout
+	createRequest.MaxParticipants = conf.MaxParticipants
+	createRequest.MinPlayoutDelay = conf.MinPlayoutDelay
+	createRequest.MaxPlayoutDelay = conf.MaxPlayoutDelay
+	createRequest.SyncStreams = conf.SyncStreams
+}

pkg/sfu/buffer/buffer.go

@@ -134,7 +134,6 @@ type Buffer struct {
 	packetNotFoundCount   atomic.Uint32
 	packetTooOldCount     atomic.Uint32
 	extPacketTooMuchCount atomic.Uint32
-	invalidPacketCount    atomic.Uint32
 
 	primaryBufferForRTX *Buffer
 	rtxPktBuf           []byte
@@ -231,6 +230,10 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 		switch ext.URI {
 		case dd.ExtensionURI:
 			if IsSvcCodec(codec.MimeType) {
+				if b.ddExtID != 0 {
+					b.logger.Warnw("multiple dependency descriptor extensions found", nil, "id", ext.ID, "previous", b.ddExtID)
+					continue
+				}
 				b.ddExtID = uint8(ext.ID)
 				frc := NewFrameRateCalculatorDD(b.clockRate, b.logger)
 				for i := range b.frameRateCalculator {

pkg/sfu/rtpstats/rtpstats_sender.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/livekit/mediatransportutil"
 	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/utils/mono"
 )
 
 const (
@@ -137,6 +138,15 @@ type senderSnapshot struct {
 	intervalStats intervalStats
 }
 
+// -------------------------------------------------------------------
+
+type rttMarker struct {
+	ntpTime mediatransportutil.NtpTime
+	sentAt  time.Time
+}
+
+// -------------------------------------------------------------------
+
 type RTPStatsSender struct {
 	*rtpStatsBase
 
@@ -144,6 +154,8 @@ type RTPStatsSender struct {
 	extHighestSN       uint64
 	extHighestSNFromRR uint64
 
+	rttMarker rttMarker
+
 	lastRRTime time.Time
 	lastRR     rtcp.ReceptionReport
 
@@ -190,6 +202,8 @@ func (r *RTPStatsSender) Seed(from *RTPStatsSender) {
 	r.extHighestSN = from.extHighestSN
 	r.extHighestSNFromRR = from.extHighestSNFromRR
 
+	r.rttMarker = from.rttMarker
+
 	r.lastRRTime = from.lastRRTime
 	r.lastRR = from.lastRR
 
@@ -493,7 +507,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
 
 	if r.srNewest != nil {
 		var err error
-		rtt, err = mediatransportutil.GetRttMs(&rr, mediatransportutil.NtpTime(r.srNewest.NtpTimestamp), time.Unix(0, r.srNewest.At))
+		rtt, err = mediatransportutil.GetRttMs(&rr, r.rttMarker.ntpTime, r.rttMarker.sentAt)
 		if err == nil {
 			isRttChanged = rtt != r.rtt
 		} else {
@@ -630,17 +644,24 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek
 		return nil
 	}
 
-	timeSincePublisherSRAdjusted := time.Since(time.Unix(0, publisherSRData.AtAdjusted))
-	now := publisherSRData.AtAdjusted + timeSincePublisherSRAdjusted.Nanoseconds()
 	var (
-		nowNTP    mediatransportutil.NtpTime
-		nowRTPExt uint64
+		reportTime         int64
+		reportTimeAdjusted int64
+		nowNTP             mediatransportutil.NtpTime
+		nowRTPExt          uint64
 	)
 	if passThrough {
+		reportTime = publisherSRData.At
+		reportTimeAdjusted = publisherSRData.AtAdjusted
+
 		nowNTP = mediatransportutil.NtpTime(publisherSRData.NtpTimestamp)
 		nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset
 	} else {
-		nowNTP = mediatransportutil.ToNtpTime(time.Unix(0, now))
+		timeSincePublisherSRAdjusted := time.Since(time.Unix(0, publisherSRData.AtAdjusted))
+		reportTimeAdjusted = publisherSRData.AtAdjusted + timeSincePublisherSRAdjusted.Nanoseconds()
+		reportTime = reportTimeAdjusted
+
+		nowNTP = mediatransportutil.ToNtpTime(time.Unix(0, reportTime))
 		nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset + uint64(timeSincePublisherSRAdjusted.Nanoseconds()*int64(r.params.ClockRate)/1e9)
 	}
 
@@ -650,8 +671,8 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek
 		NtpTimestamp:    uint64(nowNTP),
 		RtpTimestamp:    uint32(nowRTPExt),
 		RtpTimestampExt: nowRTPExt,
-		At:              now,
-		AtAdjusted:      now,
+		At:              reportTime,
+		AtAdjusted:      reportTimeAdjusted,
 		Packets:         packetCount,
 		Octets:          octetCount,
 	}
@@ -661,10 +682,11 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek
 		"feed", WrappedRTCPSenderReportStateLogger{publisherSRData},
 		"tsOffset", tsOffset,
 		"timeNow", time.Now(),
-		"now", time.Unix(0, now),
-		"timeSinceHighest", time.Duration(now-r.highestTime),
-		"timeSinceFirst", time.Duration(now-r.firstTime),
-		"timeSincePublisherSRAdjusted", timeSincePublisherSRAdjusted,
+		"reportTime", time.Unix(0, reportTime),
+		"reportTimeAdjusted", time.Unix(0, reportTimeAdjusted),
+		"timeSinceHighest", time.Since(time.Unix(0, r.highestTime)),
+		"timeSinceFirst", time.Since(time.Unix(0, r.firstTime)),
+		"timeSincePublisherSRAdjusted", time.Since(time.Unix(0, publisherSRData.AtAdjusted)),
 		"timeSincePublisherSR", time.Since(time.Unix(0, publisherSRData.At)),
 		"nowRTPExt", nowRTPExt,
 		"rtpStats", lockedRTPStatsSenderLogEncoder{r},
@@ -700,6 +722,11 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek
 		r.srFirst = r.srNewest
 	}
 
+	r.rttMarker = rttMarker{
+		ntpTime: nowNTP,
+		sentAt:  mono.Now(),
+	}
+
 	return &rtcp.SenderReport{
 		SSRC:        ssrc,
 		NTPTime:     uint64(nowNTP),

test/agent_test.go

@@ -120,12 +120,10 @@ func TestAgentNamespaces(t *testing.T) {
 
 	_, err = roomClient.CreateRoom(contextWithToken(createRoomToken()), &livekit.CreateRoomRequest{
 		Name: testRoom,
-		Agent: &livekit.RoomAgent{
-			Dispatches: []*livekit.RoomAgentDispatch{
-				&livekit.RoomAgentDispatch{},
-				&livekit.RoomAgentDispatch{
-					AgentName: "ag",
-				},
+		Agents: []*livekit.RoomAgentDispatch{
+			{},
+			{
+				AgentName: "ag",
 			},
 		},
 	})