From 1ba2cadccdb4c23b166f671ef909738e01ccb641 Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Fri, 25 Sep 2026 00:58:22 -0700 Subject: [PATCH] agent: release JobTerminate handler when job ends before registration (#4908) * agent: release JobTerminate handler when job ends before registration JobRequest registered the job's JobTerminate handler after AssignJob returned, so an ended UpdateJobStatus or worker disconnect handled in between deregistered nothing and the handler leaked. AssignJob also sent the assignment before recording the job as running, so an immediate ended update was dropped. Record the job before sending the assignment, and after registering the handler recheck under h.mu that the job is still running and its worker still registered, deregistering otherwise. Fixes #4901 * replace utils.CloneProto with proto.CloneOf * agent: wait on worker readiness and poll for released handlers in test --- pkg/agent/agent_test.go | 110 ++++++++++++++++++ pkg/agent/worker.go | 21 ++-- .../staticconfiguration.go | 3 +- pkg/routing/node.go | 8 +- pkg/routing/nodestats.go | 5 +- pkg/rtc/datatrack.go | 4 +- pkg/rtc/mediatrackreceiver.go | 15 ++- pkg/rtc/participant.go | 36 +++--- pkg/rtc/participant_data_track.go | 18 +-- pkg/rtc/participant_internal_test.go | 2 +- pkg/rtc/participant_sdp.go | 4 +- pkg/rtc/room.go | 8 +- pkg/rtc/room_test.go | 15 ++- pkg/rtc/subscribedtrack.go | 2 +- pkg/rtc/transportmanager.go | 2 +- pkg/service/agent_dispatch_service.go | 5 +- pkg/service/agentservice.go | 17 ++- pkg/service/localstore.go | 11 +- pkg/service/redisstore.go | 6 +- pkg/service/redisstore_test.go | 6 +- pkg/service/roomallocator.go | 8 +- pkg/sfu/forwarder.go | 6 +- pkg/sfu/receiver_base.go | 9 +- pkg/sfu/rtpstats/rtpstats_base.go | 5 +- pkg/sfu/rtpstats/rtpstats_receiver.go | 5 +- pkg/sfu/streamtrackermanager.go | 6 +- pkg/telemetry/statsworker.go | 4 +- pkg/utils/protocol.go | 5 +- 28 files changed, 237 insertions(+), 109 deletions(-) diff --git a/pkg/agent/agent_test.go b/pkg/agent/agent_test.go index 94be44aa7..0a546d7da 100644 --- a/pkg/agent/agent_test.go +++ b/pkg/agent/agent_test.go @@ -2,6 +2,7 @@ package agent_test import ( "context" + "errors" "fmt" "sync" "testing" @@ -256,3 +257,112 @@ func TestDrainConnectionsDoesNotDeadlock(t *testing.T) { }) } } + +func TestJobTerminateHandlerReleased(t *testing.T) { + const jobCount = 50 + + requestJobs := func(t *testing.T, client rpc.AgentInternalClient, agentName func(i int) string) []string { + jobIDs := make([]string, jobCount) + errs := make([]error, jobCount) + var wg sync.WaitGroup + for i := range jobCount { + job := &livekit.Job{ + Id: guid.New(guid.AgentJobPrefix), + DispatchId: guid.New(guid.AgentDispatchPrefix), + Type: livekit.JobType_JT_ROOM, + Room: &livekit.Room{}, + AgentName: agentName(i), + } + jobIDs[i] = job.Id + wg.Add(1) + go func() { + defer wg.Done() + _, errs[i] = client.JobRequest(context.Background(), job.AgentName, agent.RoomAgentTopic, job) + }() + } + wg.Wait() + for _, err := range errs { + require.NoError(t, err) + } + return jobIDs + } + + // the server reads the ping only after registering the worker's job request topic. + waitRegistered := func(t *testing.T, w *testutils.AgentWorker, agentName string) { + pongs := w.WorkerPongs.Observe() + defer pongs.Stop() + w.Register(agentName, livekit.JobType_JT_ROOM) + w.SendPing(&livekit.WorkerPing{Timestamp: time.Now().UnixMilli()}) + select { + case <-pongs.Events(): + case <-time.After(5 * time.Second): + require.Fail(t, "registration timeout") + } + } + + // any response means a JobTerminate handler is still registered. + countAnswered := func(client rpc.AgentInternalClient, jobIDs []string) int32 { + var answered atomic.Int32 + var wg sync.WaitGroup + for _, id := range jobIDs { + wg.Add(1) + go func() { + defer wg.Done() + _, err := client.JobTerminate(context.Background(), id, &rpc.JobTerminateRequest{JobId: id}, psrpc.WithRequestTimeout(200*time.Millisecond)) + if !errors.Is(err, psrpc.ErrRequestTimedOut) && !errors.Is(err, psrpc.ErrNoResponse) { + answered.Inc() + } + }() + } + wg.Wait() + return answered.Load() + } + + // handlers are released as the server processes the worker's messages; a leaked one never is. + requireHandlersReleased := func(t *testing.T, client rpc.AgentInternalClient, jobIDs []string) { + var answered int32 + require.Eventually(t, func() bool { + answered = countAnswered(client, jobIDs) + return answered == 0 + }, 10*time.Second, 100*time.Millisecond, "jobs with leaked JobTerminate handlers: %d", answered) + } + + t.Run("job fails on assignment", func(t *testing.T) { + bus := psrpc.NewLocalMessageBus() + client := must.Get(rpc.NewAgentInternalClient(bus)) + server := testutils.NewTestServer(bus) + t.Cleanup(server.Close) + + var worker *testutils.AgentWorker + worker = server.SimulateAgentWorker(testutils.WithJobAssignmentHandler(func(j *livekit.Job) testutils.JobLoad { + worker.SendUpdateJob(&livekit.UpdateJobStatus{JobId: j.Id, Status: livekit.JobStatus_JS_FAILED}) + return testutils.NewStableJobLoad(0) + })) + waitRegistered(t, worker, "fail_agent") + + jobIDs := requestJobs(t, client, func(int) string { return "fail_agent" }) + + requireHandlersReleased(t, client, jobIDs) + }) + + t.Run("worker disconnects on assignment", func(t *testing.T) { + bus := psrpc.NewLocalMessageBus() + client := must.Get(rpc.NewAgentInternalClient(bus)) + server := testutils.NewTestServer(bus) + t.Cleanup(server.Close) + + agentName := func(i int) string { return fmt.Sprintf("disconnect_agent_%d", i) } + for i := range jobCount { + var worker *testutils.AgentWorker + worker = server.SimulateAgentWorker(testutils.WithJobAssignmentHandler(func(j *livekit.Job) testutils.JobLoad { + worker.Close() + return nil + })) + waitRegistered(t, worker, agentName(i)) + } + + jobIDs := requestJobs(t, client, agentName) + + requireHandlersReleased(t, client, jobIDs) + }) +} diff --git a/pkg/agent/worker.go b/pkg/agent/worker.go index d8d14b94a..75f80f00f 100644 --- a/pkg/agent/worker.go +++ b/pkg/agent/worker.go @@ -29,7 +29,6 @@ import ( "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/rpc" - "github.com/livekit/protocol/utils" "github.com/livekit/protocol/utils/guid" "github.com/livekit/psrpc" ) @@ -333,12 +332,12 @@ func (w *Worker) GetJobState(jobID livekit.JobID) (*livekit.JobState, error) { if !ok { return nil, ErrJobNotFound } - return utils.CloneProto(j.State), nil + return proto.CloneOf(j.State), nil } func (w *Worker) AssignJob(ctx context.Context, job *livekit.Job, hook AssignmentHook) (*livekit.JobState, error) { availCh := make(chan *livekit.AvailabilityResponse, 1) - job = utils.CloneProto(job) + job = proto.CloneOf(job) jobID := livekit.JobID(job.Id) w.mu.Lock() @@ -420,16 +419,20 @@ func (w *Worker) AssignJob(ctx context.Context, job *livekit.Job, hook Assignmen if hook != nil { send = hook(send) } + state := proto.CloneOf(job.State) + + // must be in runningJobs before the assignment is sent; stored as a copy because send reads job without w.mu. + w.mu.Lock() + w.runningJobs[jobID] = proto.CloneOf(job) + w.mu.Unlock() + if err := send(&livekit.JobAssignment{Job: job, Token: token}); err != nil { + w.mu.Lock() + delete(w.runningJobs, jobID) + w.mu.Unlock() return nil, err } - state := utils.CloneProto(job.State) - - w.mu.Lock() - w.runningJobs[jobID] = job - w.mu.Unlock() - // TODO sweep jobs that are never started. We can't do this until all SDKs actually update the the JOB state return state, nil diff --git a/pkg/clientconfiguration/staticconfiguration.go b/pkg/clientconfiguration/staticconfiguration.go index d8145c260..198b81dc9 100644 --- a/pkg/clientconfiguration/staticconfiguration.go +++ b/pkg/clientconfiguration/staticconfiguration.go @@ -20,7 +20,6 @@ import ( "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - protoutils "github.com/livekit/protocol/utils" ) type ConfigurationItem struct { @@ -59,7 +58,7 @@ func (s *StaticClientConfigurationManager) GetConfiguration(clientInfo *livekit. var conf *livekit.ClientConfiguration for k, v := range matchedConf { if k == 0 { - conf = protoutils.CloneProto(matchedConf[0]) + conf = proto.CloneOf(matchedConf[0]) } else { // TODO : there is a problem use protobuf merge, we don't have flag to indicate 'no value', // don't override default behavior or other configuration's field. So a bool value = false or diff --git a/pkg/routing/node.go b/pkg/routing/node.go index df47226e9..dceddb92b 100644 --- a/pkg/routing/node.go +++ b/pkg/routing/node.go @@ -19,6 +19,8 @@ import ( "sync" "time" + "google.golang.org/protobuf/proto" + "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/utils" "github.com/livekit/protocol/utils/guid" @@ -76,14 +78,14 @@ func NewLocalNode(conf *config.Config) (*LocalNodeImpl, error) { } func NewLocalNodeFromNodeProto(node *livekit.Node) (*LocalNodeImpl, error) { - return &LocalNodeImpl{node: utils.CloneProto(node)}, nil + return &LocalNodeImpl{node: proto.CloneOf(node)}, nil } func (l *LocalNodeImpl) Clone() *livekit.Node { l.lock.RLock() defer l.lock.RUnlock() - return utils.CloneProto(l.node) + return proto.CloneOf(l.node) } // for testing only @@ -134,7 +136,7 @@ func (l *LocalNodeImpl) SetStats(stats *livekit.NodeStats) { l.lock.Lock() defer l.lock.Unlock() - l.node.Stats = utils.CloneProto(stats) + l.node.Stats = proto.CloneOf(stats) } func (l *LocalNodeImpl) UpdateNodeStats() bool { diff --git a/pkg/routing/nodestats.go b/pkg/routing/nodestats.go index ac16bb0ef..1a49a01af 100644 --- a/pkg/routing/nodestats.go +++ b/pkg/routing/nodestats.go @@ -18,9 +18,10 @@ import ( "sync" "time" + "google.golang.org/protobuf/proto" + "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" @@ -87,5 +88,5 @@ func (n *NodeStats) GetLatestNodeStats() *livekit.NodeStats { defer n.lock.RUnlock() latestPtr := (n.statsHistoryWritePtr - 1 + len(n.statsHistory)) % len(n.statsHistory) - return utils.CloneProto(n.statsHistory[latestPtr]) + return proto.CloneOf(n.statsHistory[latestPtr]) } diff --git a/pkg/rtc/datatrack.go b/pkg/rtc/datatrack.go index 1d93c296d..e5fd98a7c 100644 --- a/pkg/rtc/datatrack.go +++ b/pkg/rtc/datatrack.go @@ -20,13 +20,13 @@ import ( "sync" "github.com/frostbyte73/core" + "google.golang.org/protobuf/proto" "github.com/livekit/livekit-server/pkg/rtc/types" sfuutils "github.com/livekit/livekit-server/pkg/sfu/utils" "github.com/livekit/protocol/datatrack" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils" ) var ( @@ -98,7 +98,7 @@ func (d *DataTrack) PublisherIdentity() livekit.ParticipantIdentity { } func (d *DataTrack) ToProto() *livekit.DataTrackInfo { - return utils.CloneProto(d.dti) + return proto.CloneOf(d.dti) } func (d *DataTrack) PubHandle() uint16 { diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index c6130e352..e372207f9 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -30,7 +30,6 @@ import ( "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" @@ -154,7 +153,7 @@ func NewMediaTrackReceiver(params MediaTrackReceiverParams, ti *livekit.TrackInf params: params, state: mediaTrackReceiverStateOpen, } - t.trackInfo.Store(utils.CloneProto(ti)) + t.trackInfo.Store(proto.CloneOf(ti)) t.MediaTrackSubscriptions = NewMediaTrackSubscriptions(MediaTrackSubscriptionsParams{ MediaTrack: params.MediaTrack, @@ -812,7 +811,7 @@ func (t *MediaTrackReceiver) UpdateCodecInfo(codecs []*livekit.SimulcastCodec) { if len(c.Layers) != 0 { clonedLayers := make([]*livekit.VideoLayer, 0, len(c.Layers)) for _, l := range c.Layers { - clonedLayers = append(clonedLayers, utils.CloneProto(l)) + clonedLayers = append(clonedLayers, proto.CloneOf(l)) } origin.Layers = clonedLayers @@ -871,7 +870,7 @@ func (t *MediaTrackReceiver) UpdateCodecRids(mimeType mime.MimeType, rids buffer } func (t *MediaTrackReceiver) UpdateTrackInfo(ti *livekit.TrackInfo) { - clonedInfo := utils.CloneProto(ti) + clonedInfo := proto.CloneOf(ti) t.lock.Lock() trackInfo := t.TrackInfo() @@ -924,7 +923,7 @@ func (t *MediaTrackReceiver) UpdateAudioTrack(update *livekit.UpdateLocalAudioTr t.lock.Lock() trackInfo := t.TrackInfo() - clonedInfo := utils.CloneProto(trackInfo) + clonedInfo := proto.CloneOf(trackInfo) clonedInfo.AudioFeatures = sutils.DedupeSlice(update.Features) @@ -960,7 +959,7 @@ func (t *MediaTrackReceiver) UpdateVideoTrack(update *livekit.UpdateLocalVideoTr t.lock.Lock() trackInfo := t.TrackInfo() - clonedInfo := utils.CloneProto(trackInfo) + clonedInfo := proto.CloneOf(trackInfo) clonedInfo.Width = update.Width clonedInfo.Height = update.Height if proto.Equal(trackInfo, clonedInfo) { @@ -981,7 +980,7 @@ func (t *MediaTrackReceiver) UpdateVideoSize(mimeType mime.MimeType, sizes []cod var changed bool t.lock.Lock() trackInfo := t.TrackInfo() - clonedInfo := utils.CloneProto(trackInfo) + clonedInfo := proto.CloneOf(trackInfo) var maxWidth, maxHeight uint32 for _, size := range sizes { if size.Width > maxWidth { @@ -1028,7 +1027,7 @@ func (t *MediaTrackReceiver) TrackInfo() *livekit.TrackInfo { } func (t *MediaTrackReceiver) TrackInfoClone() *livekit.TrackInfo { - return utils.CloneProto(t.TrackInfo()) + return proto.CloneOf(t.TrackInfo()) } func (t *MediaTrackReceiver) NotifyMaxLayerChange(mimeType mime.MimeType, maxLayer int32) { diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 2bcff98a4..af3f8e2d3 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -667,7 +667,7 @@ func (p *ParticipantImpl) GetClientInfo() *livekit.ClientInfo { func (p *ParticipantImpl) GetClientConfiguration() *livekit.ClientConfiguration { p.lock.RLock() defer p.lock.RUnlock() - return utils.CloneProto(p.params.ClientConf) + return proto.CloneOf(p.params.ClientConf) } func (p *ParticipantImpl) GetBufferFactory() *buffer.Factory { @@ -709,7 +709,7 @@ func (p *ParticipantImpl) UpdateMetadata(update *livekit.UpdateParticipantMetada sendRequestResponse := func() error { if !fromAdmin || (update.RequestId != 0 || err != nil) { requestResponse.Request = &livekit.RequestResponse_UpdateMetadata{ - UpdateMetadata: utils.CloneProto(update), + UpdateMetadata: proto.CloneOf(update), } p.sendRequestResponse(requestResponse) } @@ -985,7 +985,7 @@ func (p *ParticipantImpl) ToProtoWithVersion() (*livekit.ParticipantInfo, utils. } if !found { - pi.Tracks = append(pi.Tracks, utils.CloneProto(pti.trackInfos[0])) + pi.Tracks = append(pi.Tracks, proto.CloneOf(pti.trackInfos[0])) } } @@ -1195,7 +1195,7 @@ func (p *ParticipantImpl) HandleICETrickle(trickleRequest *livekit.TrickleReques Reason: livekit.RequestResponse_UNCLASSIFIED_ERROR, Message: err.Error(), Request: &livekit.RequestResponse_Trickle{ - Trickle: utils.CloneProto(trickleRequest), + Trickle: proto.CloneOf(trickleRequest), }, }) return @@ -1361,7 +1361,7 @@ func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) { p.sendRequestResponse(&livekit.RequestResponse{ Reason: livekit.RequestResponse_NOT_ALLOWED, Request: &livekit.RequestResponse_AddTrack{ - AddTrack: utils.CloneProto(req), + AddTrack: proto.CloneOf(req), }, }) return @@ -1372,7 +1372,7 @@ func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) { p.sendRequestResponse(&livekit.RequestResponse{ Reason: livekit.RequestResponse_UNSUPPORTED_TYPE, Request: &livekit.RequestResponse_AddTrack{ - AddTrack: utils.CloneProto(req), + AddTrack: proto.CloneOf(req), }, }) return @@ -2948,7 +2948,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit p.sendRequestResponse(&livekit.RequestResponse{ Reason: livekit.RequestResponse_NOT_FOUND, Request: &livekit.RequestResponse_AddTrack{ - AddTrack: utils.CloneProto(req), + AddTrack: proto.CloneOf(req), }, }) p.pendingTracksLock.Unlock() @@ -2967,7 +2967,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit Reason: livekit.RequestResponse_LIMIT_EXCEEDED, Message: fmt.Sprintf("too many pending tracks, rejecting new track, numPendingTracks: %d", len(p.pendingTracks)), Request: &livekit.RequestResponse_AddTrack{ - AddTrack: utils.CloneProto(req), + AddTrack: proto.CloneOf(req), }, }) p.pendingTracksLock.Unlock() @@ -2989,7 +2989,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit clonedLayers := make([]*livekit.VideoLayer, 0, len(layers)) for _, l := range layers { - clonedLayers = append(clonedLayers, utils.CloneProto(l)) + clonedLayers = append(clonedLayers, proto.CloneOf(l)) } slices.SortFunc(clonedLayers, func(i, j *livekit.VideoLayer) int { return int(i.Quality) - int(j.Quality) @@ -3157,7 +3157,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit Reason: livekit.RequestResponse_LIMIT_EXCEEDED, Message: fmt.Sprintf("too many pending queued tracks, rejecting new track, numPendingTracksQueued: %d", len(p.pendingTracks[req.Cid].trackInfos)), Request: &livekit.RequestResponse_AddTrack{ - AddTrack: utils.CloneProto(req), + AddTrack: proto.CloneOf(req), }, }) p.pendingTracksLock.Unlock() @@ -3175,7 +3175,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit p.sendRequestResponse(&livekit.RequestResponse{ Reason: livekit.RequestResponse_QUEUED, Request: &livekit.RequestResponse_AddTrack{ - AddTrack: utils.CloneProto(req), + AddTrack: proto.CloneOf(req), }, }) @@ -3185,7 +3185,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit } p.pendingTracksLock.Unlock() - p.GetTelemetryListener().OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti), true) + p.GetTelemetryListener().OnTrackPublishRequested(p.ID(), p.Identity(), proto.CloneOf(ti), true) return nil } @@ -3210,7 +3210,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit } p.pendingTracksLock.Unlock() - p.GetTelemetryListener().OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti), true) + p.GetTelemetryListener().OnTrackPublishRequested(p.ID(), p.Identity(), proto.CloneOf(ti), true) return ti } @@ -3262,7 +3262,7 @@ func (p *ParticipantImpl) setTrackMuted(mute *livekit.MuteTrackRequest, fromAdmi for _, pti := range p.pendingTracks { for i, ti := range pti.trackInfos { if livekit.TrackID(ti.Sid) == trackID { - ti = utils.CloneProto(ti) + ti = proto.CloneOf(ti) changed = changed || ti.Muted != mute.Muted ti.Muted = mute.Muted pti.trackInfos[i] = ti @@ -3286,7 +3286,7 @@ func (p *ParticipantImpl) setTrackMuted(mute *livekit.MuteTrackRequest, fromAdmi p.sendRequestResponse(&livekit.RequestResponse{ Reason: livekit.RequestResponse_NOT_FOUND, Request: &livekit.RequestResponse_Mute{ - Mute: utils.CloneProto(mute), + Mute: proto.CloneOf(mute), }, }) } @@ -3714,7 +3714,7 @@ func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackTyp return signalCid, nil, buffer.VideoLayersRid{}, false, time.Time{} } - return signalCid, utils.CloneProto(pendingInfo.trackInfos[0]), pendingInfo.sdpRids, pendingInfo.migrated, pendingInfo.createdAt + return signalCid, proto.CloneOf(pendingInfo.trackInfos[0]), pendingInfo.sdpRids, pendingInfo.migrated, pendingInfo.createdAt } func (p *ParticipantImpl) getPendingTrackPrimaryBySdpCid(sdpCid string) *pendingTrackInfo { @@ -4231,7 +4231,7 @@ func (p *ParticipantImpl) UpdateAudioTrack(update *livekit.UpdateLocalAudioTrack p.sendRequestResponse(&livekit.RequestResponse{ Reason: livekit.RequestResponse_NOT_FOUND, Request: &livekit.RequestResponse_UpdateAudioTrack{ - UpdateAudioTrack: utils.CloneProto(update), + UpdateAudioTrack: proto.CloneOf(update), }, }) return errors.New("could not find track") @@ -4263,7 +4263,7 @@ func (p *ParticipantImpl) UpdateVideoTrack(update *livekit.UpdateLocalVideoTrack p.sendRequestResponse(&livekit.RequestResponse{ Reason: livekit.RequestResponse_NOT_FOUND, Request: &livekit.RequestResponse_UpdateVideoTrack{ - UpdateVideoTrack: utils.CloneProto(update), + UpdateVideoTrack: proto.CloneOf(update), }, }) return errors.New("could not find track") diff --git a/pkg/rtc/participant_data_track.go b/pkg/rtc/participant_data_track.go index 23a34134f..f025c6452 100644 --- a/pkg/rtc/participant_data_track.go +++ b/pkg/rtc/participant_data_track.go @@ -15,6 +15,8 @@ package rtc import ( + "google.golang.org/protobuf/proto" + "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/protocol/datatrack" "github.com/livekit/protocol/livekit" @@ -30,7 +32,7 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData Reason: livekit.RequestResponse_NOT_ALLOWED, Message: "does not have permission to publish data", Request: &livekit.RequestResponse_PublishDataTrack{ - PublishDataTrack: utils.CloneProto(req), + PublishDataTrack: proto.CloneOf(req), }, }) return @@ -42,7 +44,7 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData Reason: livekit.RequestResponse_INVALID_HANDLE, Message: "handle should be > 0 AND < 65536", Request: &livekit.RequestResponse_PublishDataTrack{ - PublishDataTrack: utils.CloneProto(req), + PublishDataTrack: proto.CloneOf(req), }, }) return @@ -54,7 +56,7 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData Reason: livekit.RequestResponse_INVALID_NAME, Message: "name should not be empty and should not exceed 256 characters", Request: &livekit.RequestResponse_PublishDataTrack{ - PublishDataTrack: utils.CloneProto(req), + PublishDataTrack: proto.CloneOf(req), }, }) return @@ -67,7 +69,7 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData Reason: livekit.RequestResponse_INVALID_REQUEST, Message: "encoding identifier is empty or exceeds the maximum length", Request: &livekit.RequestResponse_PublishDataTrack{ - PublishDataTrack: utils.CloneProto(req), + PublishDataTrack: proto.CloneOf(req), }, }) return @@ -95,7 +97,7 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData Reason: reason, Message: message, Request: &livekit.RequestResponse_PublishDataTrack{ - PublishDataTrack: utils.CloneProto(req), + PublishDataTrack: proto.CloneOf(req), }, }) return @@ -108,8 +110,8 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData Name: req.Name, Encryption: req.Encryption, } - dti.FrameEncoding = utils.CloneProto(req.GetFrameEncoding()) - dti.Schema = utils.CloneProto(req.GetSchema()) + dti.FrameEncoding = proto.CloneOf(req.GetFrameEncoding()) + dti.Schema = proto.CloneOf(req.GetSchema()) dt := NewDataTrack( DataTrackParams{ Logger: p.params.Logger.WithValues("trackID", dti.Sid), @@ -143,7 +145,7 @@ func (p *ParticipantImpl) HandleUnpublishDataTrackRequest(req *livekit.Unpublish p.sendRequestResponse(&livekit.RequestResponse{ Reason: livekit.RequestResponse_NOT_FOUND, Request: &livekit.RequestResponse_UnpublishDataTrack{ - UnpublishDataTrack: utils.CloneProto(req), + UnpublishDataTrack: proto.CloneOf(req), }, }) return diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 3a32808ac..da41e7ff0 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -560,7 +560,7 @@ func TestPreferMediaCodecForPublisher(t *testing.T) { for i := 0; i < 2; i++ { // publish preferred track without client using setCodecPreferences() trackCid := fmt.Sprintf("%s-%d", tc.trackBaseCid, i) - req := utils.CloneProto(tc.addTrack) + req := proto.CloneOf(tc.addTrack) req.SimulcastCodecs = []*livekit.SimulcastCodec{ { Codec: tc.preferredCodec, diff --git a/pkg/rtc/participant_sdp.go b/pkg/rtc/participant_sdp.go index ca8ae095a..5896e99df 100644 --- a/pkg/rtc/participant_sdp.go +++ b/pkg/rtc/participant_sdp.go @@ -22,13 +22,13 @@ import ( "github.com/pion/sdp/v3" "github.com/pion/webrtc/v4" + "google.golang.org/protobuf/proto" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" lksdp "github.com/livekit/protocol/sdp" - "github.com/livekit/protocol/utils" ) func (p *ParticipantImpl) populateSdpCid(parsedOffer *sdp.SessionDescription) ([]*sdp.MediaDescription, []*sdp.MediaDescription) { @@ -143,7 +143,7 @@ func (p *ParticipantImpl) populateSdpCid(parsedOffer *sdp.SessionDescription) ([ } if updated { - p.pendingTracks[signalCid].trackInfos[0] = utils.CloneProto(info) + p.pendingTracks[signalCid].trackInfos[0] = proto.CloneOf(info) p.pubLogger.Debugw( "pending track SDP cid updated", "signalCid", signalCid, diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 76224e731..3ae984632 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -254,7 +254,7 @@ func NewRoom( egressLauncher EgressLauncher, ) *Room { r := &Room{ - protoRoom: utils.CloneProto(room), + protoRoom: proto.CloneOf(room), internal: internal, logger: LoggerWithRoom( logger.GetLogger().WithComponent(sutils.ComponentRoom), @@ -874,7 +874,7 @@ func (r *Room) GetAgentDispatches(dispatchID string) ([]*livekit.AgentDispatch, for _, ad := range r.agentDispatches { if dispatchID == "" || ad.Id == dispatchID { - ret = append(ret, utils.CloneProto(ad.AgentDispatch)) + ret = append(ret, proto.CloneOf(ad.AgentDispatch)) } } @@ -1587,7 +1587,7 @@ func (r *Room) sendSpeakerChanges(speakers []*livekit.SpeakerInfo) { func (r *Room) updateProto() *livekit.Room { r.lock.RLock() - room := utils.CloneProto(r.protoRoom) + room := proto.CloneOf(r.protoRoom) r.lock.RUnlock() room.NumPublishers = 0 @@ -1658,7 +1658,7 @@ func (r *Room) audioUpdateWorker() { // changedSpeakers need to include previous speakers that are no longer speaking for sid, speaker := range lastActiveMap { if nextActiveMap[sid] == nil { - inactiveSpeaker := utils.CloneProto(speaker) + inactiveSpeaker := proto.CloneOf(speaker) inactiveSpeaker.Level = 0 inactiveSpeaker.Active = false changedSpeakers = append(changedSpeakers, inactiveSpeaker) diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index 8ccd9574f..37055c738 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -24,7 +24,6 @@ import ( "github.com/livekit/protocol/auth/authfakes" "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/utils" "github.com/livekit/protocol/webhook" "github.com/livekit/livekit-server/version" @@ -291,7 +290,7 @@ func TestPushAndDequeueUpdates(t *testing.T) { { name: "last version is enqueued", pi: subscriber1v2, - existing: &ParticipantUpdate{ParticipantInfo: utils.CloneProto(subscriber1v1)}, // clone the existing value since it can be modified when setting to disconnected + existing: &ParticipantUpdate{ParticipantInfo: proto.CloneOf(subscriber1v1)}, // clone the existing value since it can be modified when setting to disconnected validate: func(t *testing.T, rm *Room, _ []*ParticipantUpdate) { queued := rm.batchedUpdates[livekit.ParticipantIdentity(identity)] require.NotNil(t, queued) @@ -301,7 +300,7 @@ func TestPushAndDequeueUpdates(t *testing.T) { { name: "latest version when immediate", pi: subscriber1v2, - existing: &ParticipantUpdate{ParticipantInfo: utils.CloneProto(subscriber1v1)}, + existing: &ParticipantUpdate{ParticipantInfo: proto.CloneOf(subscriber1v1)}, immediate: true, expected: []*ParticipantUpdate{{ParticipantInfo: subscriber1v2}}, validate: func(t *testing.T, rm *Room, _ []*ParticipantUpdate) { @@ -312,7 +311,7 @@ func TestPushAndDequeueUpdates(t *testing.T) { { name: "out of order updates are rejected", pi: subscriber1v1, - existing: &ParticipantUpdate{ParticipantInfo: utils.CloneProto(subscriber1v2)}, + existing: &ParticipantUpdate{ParticipantInfo: proto.CloneOf(subscriber1v2)}, validate: func(t *testing.T, rm *Room, updates []*ParticipantUpdate) { queued := rm.batchedUpdates[livekit.ParticipantIdentity(identity)] requirePIEquals(t, subscriber1v2, queued.ParticipantInfo) @@ -322,7 +321,7 @@ func TestPushAndDequeueUpdates(t *testing.T) { name: "sid change is broadcasted immediately with synthsized disconnect", pi: publisher2, closeReason: types.ParticipantCloseReasonServiceRequestRemoveParticipant, // just to test if update contain the close reason - existing: &ParticipantUpdate{ParticipantInfo: utils.CloneProto(subscriber1v2), CloseReason: types.ParticipantCloseReasonStale}, + existing: &ParticipantUpdate{ParticipantInfo: proto.CloneOf(subscriber1v2), CloseReason: types.ParticipantCloseReasonStale}, expected: []*ParticipantUpdate{ { ParticipantInfo: &livekit.ParticipantInfo{ @@ -340,7 +339,7 @@ func TestPushAndDequeueUpdates(t *testing.T) { { name: "when switching to publisher, queue is cleared", pi: publisher1v2, - existing: &ParticipantUpdate{ParticipantInfo: utils.CloneProto(subscriber1v1)}, + existing: &ParticipantUpdate{ParticipantInfo: proto.CloneOf(subscriber1v1)}, expected: []*ParticipantUpdate{{ParticipantInfo: publisher1v2}}, validate: func(t *testing.T, rm *Room, updates []*ParticipantUpdate) { require.Empty(t, rm.batchedUpdates) @@ -651,7 +650,7 @@ func TestDataChannel(t *testing.T) { } setSource(mode, packet, p) - packetExp := utils.CloneProto(packet) + packetExp := proto.CloneOf(packet) if mode != legacySID { packetExp.ParticipantIdentity = string(p.Identity()) packetExp.GetUser().ParticipantIdentity = string(p.Identity()) @@ -699,7 +698,7 @@ func TestDataChannel(t *testing.T) { setSource(mode, packet, p) setDest(mode, packet, p1) - packetExp := utils.CloneProto(packet) + packetExp := proto.CloneOf(packet) if mode != legacySID { packetExp.ParticipantIdentity = string(p.Identity()) packetExp.GetUser().ParticipantIdentity = string(p.Identity()) diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 27d18c382..1771767d6 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -323,7 +323,7 @@ func (t *SubscribedTrack) UpdateSubscriberSettings(settings *livekit.UpdateTrack } isImmediate = isImmediate || (!settings.Disabled && settings.Disabled != t.isMutedLocked()) - t.settings = utils.CloneProto(settings) + t.settings = proto.CloneOf(settings) t.logger.Debugw("saving subscriber track settings", "settings", logger.Proto(t.settings)) t.settingsLock.Unlock() diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 2fb5c22e7..e6d983a43 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -675,7 +675,7 @@ func (t *TransportManager) GetICEConfig() *livekit.ICEConfig { if t.iceConfig == nil { return nil } - return utils.CloneProto(t.iceConfig) + return proto.CloneOf(t.iceConfig) } func (t *TransportManager) resetTransportConfigureLocked(reconfigured bool) { diff --git a/pkg/service/agent_dispatch_service.go b/pkg/service/agent_dispatch_service.go index f2dc581f5..bab17b268 100644 --- a/pkg/service/agent_dispatch_service.go +++ b/pkg/service/agent_dispatch_service.go @@ -18,13 +18,14 @@ import ( "context" "fmt" + "google.golang.org/protobuf/proto" + "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/protocol/agent" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/rpc" - "github.com/livekit/protocol/utils" "github.com/livekit/protocol/utils/guid" "github.com/livekit/psrpc" ) @@ -120,7 +121,7 @@ func redactCreateAgentDispatchRequest(req *livekit.CreateAgentDispatchRequest) * return req } - clone := utils.CloneProto(req) + clone := proto.CloneOf(req) // replace with size of metadata to provide visibility on request size if clone.Metadata != "" { diff --git a/pkg/service/agentservice.go b/pkg/service/agentservice.go index b5128cb9e..3e58e832b 100644 --- a/pkg/service/agentservice.go +++ b/pkg/service/agentservice.go @@ -423,14 +423,19 @@ func (h *AgentHandler) JobRequest(ctx context.Context, job *livekit.Job) (*rpc.J switch state.GetStatus() { case livekit.JobStatus_JS_RUNNING: logger.Infow("assigned job to worker", "apiKey", selected.APIKey()) - h.mu.Lock() - h.jobToWorker[livekit.JobID(job.Id)] = selected - h.mu.Unlock() - - err = h.agentServer.RegisterJobTerminateTopic(job.Id) - if err != nil { + jobID := livekit.JobID(job.Id) + if err := h.agentServer.RegisterJobTerminateTopic(job.Id); err != nil { logger.Errorw("failed to register JobTerminate handler", err) } + + // checked under h.mu after registering so that either this or the job/worker cleanup, whichever runs second, releases the handler. + h.mu.Lock() + if _, err := selected.GetJobState(jobID); err == nil && h.workers[selected.ID] == selected { + h.jobToWorker[jobID] = selected + } else { + h.deregisterJob(jobID) + } + h.mu.Unlock() fallthrough case livekit.JobStatus_JS_SUCCESS: return &rpc.JobRequestResponse{ diff --git a/pkg/service/localstore.go b/pkg/service/localstore.go index 1cf747f94..863a90ae0 100644 --- a/pkg/service/localstore.go +++ b/pkg/service/localstore.go @@ -20,6 +20,7 @@ import ( "time" "github.com/thoas/go-funk" + "google.golang.org/protobuf/proto" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/utils" @@ -203,7 +204,7 @@ func (s *LocalStore) StoreAgentDispatch(ctx context.Context, dispatch *livekit.A s.lock.Lock() defer s.lock.Unlock() - clone := utils.CloneProto(dispatch) + clone := proto.CloneOf(dispatch) if clone.State != nil { clone.State.Jobs = nil } @@ -242,13 +243,13 @@ func (s *LocalStore) ListAgentDispatches(ctx context.Context, roomName livekit.R var js []*livekit.Job for _, j := range agentJobs { - js = append(js, utils.CloneProto(j)) + js = append(js, proto.CloneOf(j)) } var ds []*livekit.AgentDispatch m := make(map[string]*livekit.AgentDispatch) for _, d := range agentDispatches { - clone := utils.CloneProto(d) + clone := proto.CloneOf(d) m[d.Id] = clone ds = append(ds, clone) } @@ -256,7 +257,7 @@ func (s *LocalStore) ListAgentDispatches(ctx context.Context, roomName livekit.R for _, j := range js { d := m[j.DispatchId] if d != nil { - d.State.Jobs = append(d.State.Jobs, utils.CloneProto(j)) + d.State.Jobs = append(d.State.Jobs, proto.CloneOf(j)) } } @@ -267,7 +268,7 @@ func (s *LocalStore) StoreAgentJob(ctx context.Context, job *livekit.Job) error s.lock.Lock() defer s.lock.Unlock() - clone := utils.CloneProto(job) + clone := proto.CloneOf(job) clone.Room = nil if clone.Participant != nil { clone.Participant = &livekit.ParticipantInfo{ diff --git a/pkg/service/redisstore.go b/pkg/service/redisstore.go index 292ed9524..936f403a1 100644 --- a/pkg/service/redisstore.go +++ b/pkg/service/redisstore.go @@ -548,7 +548,7 @@ func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo) } // ignore state - infoCopy := utils.CloneProto(info) + infoCopy := proto.CloneOf(info) infoCopy.State = nil data, err := proto.Marshal(infoCopy) @@ -851,7 +851,7 @@ func (s *RedisStore) DeleteIngress(_ context.Context, info *livekit.IngressInfo) } func (s *RedisStore) StoreAgentDispatch(_ context.Context, dispatch *livekit.AgentDispatch) error { - di := utils.CloneProto(dispatch) + di := proto.CloneOf(dispatch) // Do not store jobs with the dispatch if di.State != nil { @@ -914,7 +914,7 @@ func (s *RedisStore) StoreAgentJob(_ context.Context, job *livekit.Job) error { key := AgentJobPrefix + string(job.Room.Name) - jb := utils.CloneProto(job) + jb := proto.CloneOf(job) // Do not store room with the job jb.Room = nil diff --git a/pkg/service/redisstore_test.go b/pkg/service/redisstore_test.go index e84a6c41f..419694537 100644 --- a/pkg/service/redisstore_test.go +++ b/pkg/service/redisstore_test.go @@ -347,7 +347,7 @@ func TestAgentStore(t *testing.T) { require.NoError(t, err) require.Equal(t, 1, len(rd)) - expected := utils.CloneProto(ad) + expected := proto.CloneOf(ad) expected.State.Jobs = nil require.True(t, proto.Equal(expected, rd[0])) @@ -358,7 +358,7 @@ func TestAgentStore(t *testing.T) { require.NoError(t, err) require.Equal(t, 1, len(rd)) - expected = utils.CloneProto(ad) + expected = proto.CloneOf(ad) expected.State.Jobs[0].Room = nil expected.State.Jobs[0].Participant = &livekit.ParticipantInfo{ Identity: "identity", @@ -372,7 +372,7 @@ func TestAgentStore(t *testing.T) { require.NoError(t, err) require.Equal(t, 1, len(rd)) - expected = utils.CloneProto(ad) + expected = proto.CloneOf(ad) expected.State.Jobs = nil require.True(t, proto.Equal(expected, rd[0])) diff --git a/pkg/service/roomallocator.go b/pkg/service/roomallocator.go index 436a433c0..d843529c9 100644 --- a/pkg/service/roomallocator.go +++ b/pkg/service/roomallocator.go @@ -19,6 +19,8 @@ import ( "errors" "time" + "google.golang.org/protobuf/proto" + "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" @@ -211,7 +213,7 @@ func (r *StandardRoomAllocator) applyNamedRoomConfiguration(req *livekit.CreateR return req, psrpc.NewErrorf(psrpc.InvalidArgument, "unknown room configuration in create room request") } - clone := utils.CloneProto(req) + clone := proto.CloneOf(req) if clone.EmptyTimeout == 0 { clone.EmptyTimeout = conf.EmptyTimeout @@ -223,12 +225,12 @@ func (r *StandardRoomAllocator) applyNamedRoomConfiguration(req *livekit.CreateR clone.MaxParticipants = conf.MaxParticipants } if clone.Egress == nil { - clone.Egress = utils.CloneProto(conf.Egress) + clone.Egress = proto.CloneOf(conf.Egress) } 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)) + clone.Agents = append(clone.Agents, proto.CloneOf(agent)) } } if clone.MinPlayoutDelay == 0 { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 5dfc69038..9c87b0653 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -26,12 +26,12 @@ import ( "github.com/pion/rtp/codecs" "github.com/pion/webrtc/v4" "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/proto" "github.com/livekit/mediatransportutil" "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils" "github.com/livekit/protocol/utils/mono" "github.com/livekit/livekit-server/pkg/sfu/buffer" @@ -491,7 +491,7 @@ func (f *Forwarder) GetState() *livekit.RTPForwarderState { state.SenderReportState = make([]*livekit.RTCPSenderReportState, len(f.refInfos)) for layer, refInfo := range f.refInfos { - state.SenderReportState[layer] = utils.CloneProto(refInfo.senderReport) + state.SenderReportState[layer] = proto.CloneOf(refInfo.senderReport) } return state } @@ -506,7 +506,7 @@ func (f *Forwarder) SeedState(state *livekit.RTPForwarderState) { for layer, rtcpSenderReportState := range state.SenderReportState { f.refInfos[layer] = refInfo{} - if senderReport := utils.CloneProto(rtcpSenderReportState); senderReport != nil && senderReport.NtpTimestamp != 0 { + if senderReport := proto.CloneOf(rtcpSenderReportState); senderReport != nil && senderReport.NtpTimestamp != 0 { f.refInfos[layer].senderReport = senderReport } } diff --git a/pkg/sfu/receiver_base.go b/pkg/sfu/receiver_base.go index 678c124c3..8b3492064 100644 --- a/pkg/sfu/receiver_base.go +++ b/pkg/sfu/receiver_base.go @@ -25,6 +25,7 @@ import ( "github.com/pion/webrtc/v4" "go.uber.org/atomic" + "google.golang.org/protobuf/proto" "github.com/livekit/mediatransportutil/pkg/bucket" "github.com/livekit/mediatransportutil/pkg/codec" @@ -164,7 +165,7 @@ func NewReceiverBase(params ReceiverBaseParams, trackInfo *livekit.TrackInfo, co params: params, codecState: codecState, isRED: mime.IsMimeTypeStringRED(params.Codec.MimeType), - trackInfo: utils.CloneProto(trackInfo), + trackInfo: proto.CloneOf(trackInfo), videoLayerMode: buffer.GetVideoLayerModeForMimeType(mime.NormalizeMimeType(params.Codec.MimeType), trackInfo), } @@ -256,7 +257,7 @@ func (r *ReceiverBase) TrackInfo() *livekit.TrackInfo { r.bufferMu.RLock() defer r.bufferMu.RUnlock() - return utils.CloneProto(r.trackInfo) + return proto.CloneOf(r.trackInfo) } func (r *ReceiverBase) UpdateTrackInfo(ti *livekit.TrackInfo) { @@ -282,7 +283,7 @@ func (r *ReceiverBase) UpdateTrackInfo(ti *livekit.TrackInfo) { "shouldResync", shouldResync, ) } - r.trackInfo = utils.CloneProto(ti) + r.trackInfo = proto.CloneOf(ti) paused := r.trackInfo.GetMuted() for _, buff := range r.buffers { @@ -665,7 +666,7 @@ func (r *ReceiverBase) GetOrCreateBuffer(layer int32) buffer.BufferProvider { } r.bufferPromises[layer] = bp - ti := utils.CloneProto(r.trackInfo) + ti := proto.CloneOf(r.trackInfo) r.bufferMu.Unlock() defer close(bp.ready) diff --git a/pkg/sfu/rtpstats/rtpstats_base.go b/pkg/sfu/rtpstats/rtpstats_base.go index b5f2f6fe7..fdd92e89e 100644 --- a/pkg/sfu/rtpstats/rtpstats_base.go +++ b/pkg/sfu/rtpstats/rtpstats_base.go @@ -19,6 +19,7 @@ import ( "time" "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" "github.com/livekit/mediatransportutil" @@ -298,8 +299,8 @@ func (r *rtpStatsBase) seed(from *rtpStatsBase) bool { r.rtt = from.rtt r.maxRtt = from.maxRtt - r.srFirst = utils.CloneProto(from.srFirst) - r.srNewest = utils.CloneProto(from.srNewest) + r.srFirst = proto.CloneOf(from.srFirst) + r.srNewest = proto.CloneOf(from.srNewest) r.nextSnapshotID = from.nextSnapshotID r.snapshots = make([]snapshot, cap(from.snapshots)) diff --git a/pkg/sfu/rtpstats/rtpstats_receiver.go b/pkg/sfu/rtpstats/rtpstats_receiver.go index dfd127397..90f7eb2b2 100644 --- a/pkg/sfu/rtpstats/rtpstats_receiver.go +++ b/pkg/sfu/rtpstats/rtpstats_receiver.go @@ -22,6 +22,7 @@ import ( "github.com/pion/rtcp" "go.uber.org/zap" "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/proto" "github.com/livekit/mediatransportutil" "github.com/livekit/mediatransportutil/pkg/latency" @@ -654,7 +655,7 @@ func (r *RTPStatsReceiver) getExtendedSenderReport(srData *livekit.RTCPSenderRep } } - srDataExt := protoutils.CloneProto(srData) + srDataExt := proto.CloneOf(srData) srDataExt.RtpTimestampExt = uint64(srDataExt.RtpTimestamp) + tsCycles return srDataExt } @@ -816,7 +817,7 @@ func (r *RTPStatsReceiver) GetRtcpSenderReportData() *livekit.RTCPSenderReportSt r.lock.RLock() defer r.lock.RUnlock() - return protoutils.CloneProto(r.srNewest) + return proto.CloneOf(r.srNewest) } func (r *RTPStatsReceiver) LastSenderReportTime() time.Time { diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index 6eb28e5f2..c13ab18ad 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -20,11 +20,11 @@ import ( "time" "github.com/frostbyte73/core" + "google.golang.org/protobuf/proto" "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils" "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/streamtracker" @@ -144,7 +144,7 @@ func NewStreamTrackerManager( logger: logger, mimeType: mimeType, videoLayerMode: buffer.GetVideoLayerModeForMimeType(mimeType, trackInfo), - trackInfo: utils.CloneProto(trackInfo), + trackInfo: proto.CloneOf(trackInfo), maxPublishedLayer: buffer.InvalidLayerSpatial, maxTemporalLayerSeen: buffer.InvalidLayerTemporal, clockRate: clockRate, @@ -374,7 +374,7 @@ func (s *StreamTrackerManager) setPaused(paused bool) { func (s *StreamTrackerManager) UpdateTrackInfo(ti *livekit.TrackInfo) { s.lock.Lock() - s.trackInfo = utils.CloneProto(ti) + s.trackInfo = proto.CloneOf(ti) s.maxExpectedLayerFromTrackInfoLocked(false) paused := s.trackInfo.GetMuted() diff --git a/pkg/telemetry/statsworker.go b/pkg/telemetry/statsworker.go index 03ec21a04..463e6c1b7 100644 --- a/pkg/telemetry/statsworker.go +++ b/pkg/telemetry/statsworker.go @@ -21,12 +21,12 @@ import ( "unsafe" "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - protoutils "github.com/livekit/protocol/utils" ) type ReferenceGuard struct { @@ -419,7 +419,7 @@ func coalesce(stats []*livekit.AnalyticsStat) *livekit.AnalyticsStat { for _, videoLayer := range analyticsStream.VideoLayers { coalescedVideoLayer := coalescedVideoLayers[videoLayer.Layer] if coalescedVideoLayer == nil { - coalescedVideoLayer = protoutils.CloneProto(videoLayer) + coalescedVideoLayer = proto.CloneOf(videoLayer) coalescedVideoLayers[videoLayer.Layer] = coalescedVideoLayer } else { coalescedVideoLayer.Packets += videoLayer.Packets diff --git a/pkg/utils/protocol.go b/pkg/utils/protocol.go index 77d324b94..0cfab8aca 100644 --- a/pkg/utils/protocol.go +++ b/pkg/utils/protocol.go @@ -17,15 +17,16 @@ package utils import ( + "google.golang.org/protobuf/proto" + "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/utils" ) func ClientInfoWithoutAddress(c *livekit.ClientInfo) *livekit.ClientInfo { if c == nil { return nil } - clone := utils.CloneProto(c) + clone := proto.CloneOf(c) clone.Address = "" return clone }