diff --git a/go.mod b/go.mod index 82ff36230..c3fce73ae 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 9a3c3bb36..6d0539605 100644 --- a/go.sum +++ b/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= diff --git a/pkg/agent/testutils/server.go b/pkg/agent/testutils/server.go index 09ce9e3c9..1c4ce48f3 100644 --- a/pkg/agent/testutils/server.go +++ b/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() { diff --git a/pkg/config/config.go b/pkg/config/config.go index ccf1f68fc..463396c2a 100644 --- a/pkg/config/config.go +++ b/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 { diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 6fa66f70b..e4492bcc8 100644 --- a/pkg/routing/interfaces.go +++ b/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 diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 093c9d5f1..fcdcf68b3 100644 --- a/pkg/rtc/participant.go +++ b/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() } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 1995a3b91..7da83559f 100644 --- a/pkg/rtc/room.go +++ b/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 { diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 71c8949cc..a3665dd46 100644 --- a/pkg/rtc/transport.go +++ b/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") diff --git a/pkg/rtc/transport/handler.go b/pkg/rtc/transport/handler.go index 81ab42c03..ad70ebbb4 100644 --- a/pkg/rtc/transport/handler.go +++ b/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) diff --git a/pkg/rtc/transport/transportfakes/fake_handler.go b/pkg/rtc/transport/transportfakes/fake_handler.go index 0afd8446b..18121838c 100644 --- a/pkg/rtc/transport/transportfakes/fake_handler.go +++ b/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() { 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 { diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 40d3f1a4a..bee94e700 100644 --- a/pkg/rtc/transportmanager.go +++ b/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 } } diff --git a/pkg/rtc/types/ice.go b/pkg/rtc/types/ice.go index 3b02b4f42..96c9365dc 100644 --- a/pkg/rtc/types/ice.go +++ b/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 { diff --git a/pkg/service/auth.go b/pkg/service/auth.go index 188759fab..8edcf91c5 100644 --- a/pkg/service/auth.go +++ b/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 { diff --git a/pkg/service/roomallocator.go b/pkg/service/roomallocator.go index dce8f1d2f..e8edea3dd 100644 --- a/pkg/service/roomallocator.go +++ b/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 diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 5fd83b9d3..19b2b5737 100644 --- a/pkg/service/roommanager.go +++ b/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) diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 0d5f1bebb..5a6dafc35 100644 --- a/pkg/service/rtcservice.go +++ b/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) diff --git a/pkg/service/servicefakes/fake_session_handler.go b/pkg/service/servicefakes/fake_session_handler.go index 15d5a00d5..a31c69dbe 100644 --- a/pkg/service/servicefakes/fake_session_handler.go +++ b/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) { diff --git a/pkg/service/signal.go b/pkg/service/signal.go index 6d5523944..08d5d0a78 100644 --- a/pkg/service/signal.go +++ b/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) diff --git a/pkg/service/signal_test.go b/pkg/service/signal_test.go index 1e3ac03f2..81755cdeb 100644 --- a/pkg/service/signal_test.go +++ b/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, diff --git a/pkg/service/sip.go b/pkg/service/sip.go index 74c075669..38d45245a 100644 --- a/pkg/service/sip.go +++ b/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) diff --git a/pkg/service/utils.go b/pkg/service/utils.go index db76e008a..fd54197b2 100644 --- a/pkg/service/utils.go +++ b/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 +} diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index ea9f30eec..e0cbe66aa 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/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 { diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index c78e513e6..0c76f4352 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/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), diff --git a/test/agent_test.go b/test/agent_test.go index ad50c53d9..554e36e33 100644 --- a/test/agent_test.go +++ b/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", }, }, })