From dd7cd7eafc43cf49035bf2b9b1845b580addb99a Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 21 Oct 2024 23:31:12 -0700 Subject: [PATCH 01/12] Handle room configuration that's set in the grant itself (#3120) * Handle room configuration that's set in the grant itself * ensure refresh token contains updates * deps * dep --------- Co-authored-by: Paul Wells --- go.mod | 4 ++- go.sum | 2 -- pkg/config/config.go | 4 +-- pkg/routing/interfaces.go | 1 + pkg/rtc/room.go | 4 +-- pkg/service/auth.go | 10 ------ pkg/service/roomallocator.go | 15 ++++---- pkg/service/roommanager.go | 6 ++-- pkg/service/rtcservice.go | 11 +++--- .../servicefakes/fake_session_handler.go | 34 +++++++++---------- pkg/service/signal.go | 20 ++--------- pkg/service/signal_test.go | 2 -- pkg/service/utils.go | 15 ++++++++ test/agent_test.go | 10 +++--- 14 files changed, 65 insertions(+), 73 deletions(-) diff --git a/go.mod b/go.mod index 82ff36230..2f415c8c7 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.26.1-0.20241022053724-1eb56d424343 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..aad7823e7 100644 --- a/go.sum +++ b/go.sum @@ -165,8 +165,6 @@ 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/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/config/config.go b/pkg/config/config.go index 8225d50ac..1fceaebc9 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/room.go b/pkg/rtc/room.go index 1995a3b91..c3d643f7a 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1728,9 +1728,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 { 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/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/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", }, }, }) From b0d3d65f184607b7a54d9a2de1fc069ab1c67aee Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Mon, 21 Oct 2024 23:44:00 -0700 Subject: [PATCH 02/12] update events package (#3126) * update events package * deps --- go.mod | 2 +- go.sum | 2 ++ pkg/agent/testutils/server.go | 22 +++++++++++----------- pkg/service/sip.go | 13 ++++++------- 4 files changed, 20 insertions(+), 19 deletions(-) diff --git a/go.mod b/go.mod index 2f415c8c7..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.20241022053724-1eb56d424343 + 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 diff --git a/go.sum b/go.sum index aad7823e7..6d0539605 100644 --- a/go.sum +++ b/go.sum @@ -165,6 +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.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/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) From b30cc9066aa8e74cda16ea0f77df7094ad1a5329 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Tue, 22 Oct 2024 17:53:40 +0800 Subject: [PATCH 03/12] Drop remote candidates based on lite option (#3127) Only drop remote candidates if remote peer is not lite and local peer is lite. --- pkg/rtc/transport.go | 31 ++++++++++- pkg/rtc/transport_test.go | 110 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 1 deletion(-) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 71c8949cc..798fd8cbd 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -178,6 +178,8 @@ type PCTransport struct { firstOfferReceived bool firstOfferNoDataChannel bool + remoteICEIsLite *bool + localICEIsLite *bool reliableDC *webrtc.DataChannel reliableDCOpened bool lossyDC *webrtc.DataChannel @@ -1401,7 +1403,7 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { c := e.data.(*webrtc.ICECandidateInit) filtered := false - if t.params.DropRemoteICECandidates || (t.preferTCP.Load() && !strings.Contains(c.Candidate, "tcp")) { + if t.preferTCP.Load() && !strings.Contains(c.Candidate, "tcp") { t.params.Logger.Debugw("filtering out remote candidate", "candidate", c.Candidate) filtered = true } @@ -1421,6 +1423,11 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { return nil } + if t.params.DropRemoteICECandidates { + t.params.Logger.Debugw("dropping remote ICE candidate", "candidate", c.Candidate) + return nil + } + if err := t.pc.AddICECandidate(*c); err != nil { t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c) return errors.Wrap(err, "add ice candidate failed") @@ -1445,6 +1452,25 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, return sd } + _, iceLite := parsed.Attribute("ice-lite") + var liteSet bool + if isLocal { + if t.localICEIsLite == nil { + t.localICEIsLite = &iceLite + liteSet = true + } + } else { + if t.remoteICEIsLite == nil { + t.remoteICEIsLite = &iceLite + liteSet = true + } + } + if liteSet && t.localICEIsLite != nil && t.remoteICEIsLite != nil { + // only drop remote candidates if local is lite and remote is not + t.params.DropRemoteICECandidates = t.params.DropRemoteICECandidates && (*t.localICEIsLite && !*t.remoteICEIsLite) + t.params.Logger.Debugw("setting DropRemoteICECandidates", "dropRemoteCandidate", t.params.DropRemoteICECandidates, "localICELite", *t.localICEIsLite, "remoteICELite", *t.remoteICEIsLite) + } + filterAttributes := func(attrs []sdp.Attribute) []sdp.Attribute { filteredAttrs := make([]sdp.Attribute, 0, len(attrs)) for _, a := range attrs { @@ -1673,6 +1699,9 @@ func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error { } for _, c := range t.pendingRemoteCandidates { + if t.params.DropRemoteICECandidates { + continue + } if err := t.pc.AddICECandidate(*c); err != nil { t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c) return errors.Wrap(err, "add ice candidate failed") diff --git a/pkg/rtc/transport_test.go b/pkg/rtc/transport_test.go index c80b206ef..e475d32a3 100644 --- a/pkg/rtc/transport_test.go +++ b/pkg/rtc/transport_test.go @@ -28,6 +28,7 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/transport" "github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes" + "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/testutils" "github.com/livekit/protocol/livekit" ) @@ -503,6 +504,115 @@ func TestFilteringCandidates(t *testing.T) { transport.Close() } +func TestDropRemoteICECandidates(t *testing.T) { + cases := []struct { + name string + remoteLite bool + localLite bool + expecteLocalDrop bool + expecteRemoteDrop bool + }{ + { + name: "both not lite", + localLite: false, + remoteLite: false, + expecteLocalDrop: false, + expecteRemoteDrop: false, + }, + { + name: "remote lite", + localLite: false, + remoteLite: true, + expecteLocalDrop: false, + expecteRemoteDrop: true, + }, + { + name: "local lite", + localLite: true, + remoteLite: false, + expecteLocalDrop: true, + expecteRemoteDrop: false, + }, + { + name: "both lite", + localLite: true, + remoteLite: true, + expecteLocalDrop: false, + expecteRemoteDrop: false, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + params := TransportParams{ + ParticipantID: "id", + ParticipantIdentity: "identity", + Config: &WebRTCConfig{}, + IsOfferer: true, + ProtocolVersion: types.CurrentProtocol, + DropRemoteICECandidates: true, + } + + paramsA := params + paramsA.Config.SettingEngine.SetLite(c.localLite) + handlerA := &transportfakes.FakeHandler{} + paramsA.Handler = handlerA + transportLocal, err := NewPCTransport(paramsA) + require.NoError(t, err) + _, err = transportLocal.pc.CreateDataChannel(LossyDataChannel, nil) + require.NoError(t, err) + + paramsB := params + paramsB.Config.SettingEngine.SetLite(c.remoteLite) + handlerB := &transportfakes.FakeHandler{} + paramsB.Handler = handlerB + paramsB.IsOfferer = false + transportRemote, err := NewPCTransport(paramsB) + require.NoError(t, err) + + require.False(t, transportLocal.IsEstablished()) + require.False(t, transportRemote.IsEstablished()) + + handleICEExchange(t, transportLocal, transportRemote, handlerA, handlerB) + var offer atomic.Pointer[webrtc.SessionDescription] + handlerA.OnOfferCalls(func(sd webrtc.SessionDescription) error { + parsed, err := sd.Unmarshal() + require.NoError(t, err) + _, lite := parsed.Attribute("ice-lite") + require.Equal(t, c.localLite, lite) + offer.Store(&sd) + return nil + }) + transportLocal.Negotiate(true) + require.Eventually(t, func() bool { + return offer.Load() != nil + }, 100*time.Millisecond, time.Millisecond*10, "offer not received") + + handlerB.OnAnswerCalls(func(sd webrtc.SessionDescription) error { + parsed, err := sd.Unmarshal() + require.NoError(t, err) + _, lite := parsed.Attribute("ice-lite") + require.Equal(t, c.remoteLite, lite, sd.SDP) + transportLocal.HandleRemoteDescription(sd) + return nil + }) + transportRemote.HandleRemoteDescription(*offer.Load()) + + require.Eventually(t, func() bool { + return transportLocal.IsEstablished() + }, 10*time.Second, time.Millisecond*10, "transportA is not established") + require.Eventually(t, func() bool { + return transportRemote.IsEstablished() + }, 10*time.Second, time.Millisecond*10, "transportB is not established") + + require.Equal(t, c.expecteLocalDrop, transportLocal.params.DropRemoteICECandidates) + require.Equal(t, c.expecteRemoteDrop, transportRemote.params.DropRemoteICECandidates) + + transportLocal.Close() + transportRemote.Close() + }) + } +} func handleICEExchange(t *testing.T, a, b *PCTransport, ah, bh *transportfakes.FakeHandler) { ah.OnICECandidateCalls(func(candidate *webrtc.ICECandidate, target livekit.SignalTarget) error { From 487a3fc3fb405625e0641469513f1ec5be31b8d9 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 22 Oct 2024 20:23:55 +0530 Subject: [PATCH 04/12] ICE candidate marking (#3128) - Update filtered if dropping a pending candidate later. - Ordinal for selected pair so it is easy to see which got selected later. --- pkg/rtc/room.go | 8 +++---- pkg/rtc/transport.go | 6 +++-- pkg/rtc/types/ice.go | 52 +++++++++++++++++++++++++------------------- 3 files changed, 38 insertions(+), 28 deletions(-) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index c3d643f7a..0b19fa251 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1816,8 +1816,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]" } @@ -1829,8 +1829,8 @@ 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]" } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 798fd8cbd..a6db38c5f 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -1413,7 +1413,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 } @@ -1425,6 +1425,7 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { if t.params.DropRemoteICECandidates { t.params.Logger.Debugw("dropping remote ICE candidate", "candidate", c.Candidate) + t.connectionDetails.AddRemoteCandidate(*c, true, true, true) return nil } @@ -1494,7 +1495,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) @@ -1700,6 +1701,7 @@ 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 { 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 { From 7ab6e5df092c7fefadda293ccfb7b3f62c4ea4e5 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 23 Oct 2024 11:03:43 +0530 Subject: [PATCH 05/12] Adjust drift calculations for pass through. (#3129) No functional effect, but was logging more than expected drift in the down stream direction. Reason is that when passing through, we could be using an older report. But, the adjustment was applied to the monotonic clock and not the RTP timestamp. So, it looked like more time had elapsed for the same RTP clock elapsed and logging higher than expected drift. Correcting it so that the log is not misleading/confusing. --- pkg/sfu/rtpstats/rtpstats_sender.go | 30 ++++++++++++++++++----------- 1 file changed, 19 insertions(+), 11 deletions(-) diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index c78e513e6..e4bb542b2 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -630,17 +630,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 +657,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 +668,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}, From de102f32db94a0d050302978e702d8a4e627effb Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 23 Oct 2024 21:30:52 +0530 Subject: [PATCH 06/12] Display both pairs on selected candidate pair change (#3133) * Display both pairs on selected canddiate pair change * disable ICE lite for Firefox --- pkg/rtc/transport.go | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index a6db38c5f..fa54d3c5c 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -293,7 +293,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) @@ -473,8 +477,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) }) From ca77df8212997862199cfda60b9c8b39b211000f Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Thu, 24 Oct 2024 16:59:24 +0800 Subject: [PATCH 07/12] warn for multiple dd ext (#3135) * warn for multiple dd ext * unused --- pkg/sfu/buffer/buffer.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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 { From b8c6b1f1ecdd1b0ff136687bfecffdfa9f3bd2a2 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 24 Oct 2024 14:30:04 +0530 Subject: [PATCH 08/12] Log ICE connection info on failure. (#3134) - Truncate public remote IP - Log only on short connection to avoid logging too much --- pkg/rtc/participant.go | 2 +- pkg/rtc/room.go | 26 ++++++++++++++++- pkg/rtc/transport.go | 2 +- pkg/rtc/transport/handler.go | 3 +- .../transport/transportfakes/fake_handler.go | 19 +++++++------ pkg/rtc/transportmanager.go | 28 +++++++++++++------ 6 files changed, 60 insertions(+), 20 deletions(-) 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 0b19fa251..77471f25b 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 ( @@ -1837,7 +1840,28 @@ func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} { 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(), c.Remote.Address(), 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 fa54d3c5c..5b0111848 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -640,7 +640,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) { 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/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 } } From fbdc2491d92b06c66dc68b6ea37bb0395887ed0e Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 24 Oct 2024 16:24:54 +0530 Subject: [PATCH 09/12] Log truncated (#3136) * Log truncated * add related address --- pkg/rtc/room.go | 2 +- pkg/rtc/transport.go | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 77471f25b..a29ba6774 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1849,7 +1849,7 @@ func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} { if !isPrivate && len(remoteAddress) > minIPTruncateLen { remoteAddress = remoteAddress[:len(remoteAddress)-3] + "..." } - cStr += " " + fmt.Sprintf("%s %s %s:%d", c.Remote.NetworkType(), c.Remote.Type(), c.Remote.Address(), c.Remote.Port()) + 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 { diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 5b0111848..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 } From 024a75d27c6a0774462224039002d5468ee7eab2 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 24 Oct 2024 18:26:56 +0530 Subject: [PATCH 10/12] display related only when address is valid (#3137) --- pkg/rtc/room.go | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index a29ba6774..7da83559f 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1854,13 +1854,12 @@ func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} { 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) } - - 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) } From 542620b486c5e09865f219f6d006f008909df0f4 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 25 Oct 2024 11:11:21 +0530 Subject: [PATCH 11/12] Revert "Adjust drift calculations for pass through. (#3129)" (#3138) This reverts commit 7ab6e5df092c7fefadda293ccfb7b3f62c4ea4e5. --- pkg/sfu/rtpstats/rtpstats_sender.go | 30 +++++++++++------------------ 1 file changed, 11 insertions(+), 19 deletions(-) diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index e4bb542b2..c78e513e6 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -630,24 +630,17 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek return nil } + timeSincePublisherSRAdjusted := time.Since(time.Unix(0, publisherSRData.AtAdjusted)) + now := publisherSRData.AtAdjusted + timeSincePublisherSRAdjusted.Nanoseconds() var ( - reportTime int64 - reportTimeAdjusted int64 - nowNTP mediatransportutil.NtpTime - nowRTPExt uint64 + nowNTP mediatransportutil.NtpTime + nowRTPExt uint64 ) if passThrough { - reportTime = publisherSRData.At - reportTimeAdjusted = publisherSRData.AtAdjusted - nowNTP = mediatransportutil.NtpTime(publisherSRData.NtpTimestamp) nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset } else { - timeSincePublisherSRAdjusted := time.Since(time.Unix(0, publisherSRData.AtAdjusted)) - reportTimeAdjusted = publisherSRData.AtAdjusted + timeSincePublisherSRAdjusted.Nanoseconds() - reportTime = reportTimeAdjusted - - nowNTP = mediatransportutil.ToNtpTime(time.Unix(0, reportTime)) + nowNTP = mediatransportutil.ToNtpTime(time.Unix(0, now)) nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset + uint64(timeSincePublisherSRAdjusted.Nanoseconds()*int64(r.params.ClockRate)/1e9) } @@ -657,8 +650,8 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek NtpTimestamp: uint64(nowNTP), RtpTimestamp: uint32(nowRTPExt), RtpTimestampExt: nowRTPExt, - At: reportTime, - AtAdjusted: reportTimeAdjusted, + At: now, + AtAdjusted: now, Packets: packetCount, Octets: octetCount, } @@ -668,11 +661,10 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek "feed", WrappedRTCPSenderReportStateLogger{publisherSRData}, "tsOffset", tsOffset, "timeNow", time.Now(), - "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)), + "now", time.Unix(0, now), + "timeSinceHighest", time.Duration(now-r.highestTime), + "timeSinceFirst", time.Duration(now-r.firstTime), + "timeSincePublisherSRAdjusted", timeSincePublisherSRAdjusted, "timeSincePublisherSR", time.Since(time.Unix(0, publisherSRData.At)), "nowRTPExt", nowRTPExt, "rtpStats", lockedRTPStatsSenderLogEncoder{r}, From d341ee1ce86c943e032f249f1b1ff8950aaa0ad6 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 25 Oct 2024 11:50:59 +0530 Subject: [PATCH 12/12] Maintain RTT marker for calculations. (#3139) * Maintain RTT marker for calculations. Restore the drift logging change. * remove unnecessary cast --- pkg/sfu/rtpstats/rtpstats_sender.go | 51 ++++++++++++++++++++++------- 1 file changed, 39 insertions(+), 12 deletions(-) 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),