Use ParticipantInfo version to ensure consistency (#399)

* Use ParticipantInfo version to ensure consistency

Deprecating time.Time and avoid locking unnecessarily

* properly adjust ulimit. update protocol

* Save initial version from params

* get rid of metadata field, use grants copy

* fix test

pkg/rtc/participant.go

@@ -57,6 +57,7 @@ type ParticipantParams struct {
 	Logger                  logger.Logger
 	SimTracks               map[uint32]SimulcastTrackInfo
 	Grants                  *auth.ClaimGrants
+	InitialVersion          uint32
 }
 
 type ParticipantImpl struct {
@@ -105,6 +106,7 @@ type ParticipantImpl struct {
 	lock       sync.RWMutex
 	once       sync.Once
 	updateLock sync.Mutex
+	version    uint32
 
 	// callbacks & handlers
 	onTrackPublished func(types.LocalParticipant, types.MediaTrack)
@@ -131,6 +133,7 @@ func NewParticipant(params ParticipantParams, perms *livekit.ParticipantPermissi
 		subscribedTracksSettings: make(map[livekit.TrackID]*livekit.UpdateTrackSettings),
 		disallowedSubscriptions:  make(map[livekit.TrackID]livekit.ParticipantID),
 		connectedAt:              time.Now(),
+		version:                  params.InitialVersion,
 	}
 	p.migrateState.Store(types.MigrateStateInit)
 	p.state.Store(livekit.ParticipantInfo_JOINING)
@@ -248,8 +251,7 @@ func (p *ParticipantImpl) ConnectedAt() time.Time {
 // SetMetadata attaches metadata to the participant
 func (p *ParticipantImpl) SetMetadata(metadata string) {
 	p.lock.Lock()
-	changed := p.metadata != metadata
-	p.metadata = metadata
+	changed := p.params.Grants.Metadata != metadata
 	p.params.Grants.Metadata = metadata
 	p.lock.Unlock()
 
@@ -293,13 +295,16 @@ func (p *ParticipantImpl) ToProto() *livekit.ParticipantInfo {
 		Sid:      string(p.params.SID),
 		Identity: string(p.params.Identity),
 		Name:     string(p.params.Name),
-		Metadata: p.metadata,
 		State:    p.State(),
 		JoinedAt: p.ConnectedAt().Unix(),
 		Hidden:   p.Hidden(),
 		Recorder: p.IsRecorder(),
+		Version:  atomic.AddUint32(&p.version, 1),
 	}
 	info.Tracks = p.UpTrackManager.ToProto()
+	if p.params.Grants != nil {
+		info.Metadata = p.params.Grants.Metadata
+	}
 
 	return info
 }
@@ -594,26 +599,36 @@ func (p *ParticipantImpl) SendJoinResponse(
 	})
 }
 
-func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit.ParticipantInfo, updatedAt time.Time) error {
-	if len(participantsToUpdate) == 1 {
-		p.updateLock.Lock()
-		defer p.updateLock.Unlock()
-		pi := participantsToUpdate[0]
+func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit.ParticipantInfo) error {
+	p.updateLock.Lock()
+	validUpdates := make([]*livekit.ParticipantInfo, 0, len(participantsToUpdate))
+	for _, pi := range participantsToUpdate {
+		isValid := true
 		if val, ok := p.updateCache.Get(pi.Sid); ok {
-			if lastUpdatedAt, ok := val.(time.Time); ok {
+			if lastVersion, ok := val.(uint32); ok {
 				// this is a message delivered out of order, a more recent version of the message had already been
 				// sent.
-				if lastUpdatedAt.After(updatedAt) {
-					return nil
+				if pi.Version < lastVersion {
+					p.params.Logger.Debugw("skipping outdated participant update", "version", pi.Version, "lastVersion", lastVersion)
+					isValid = false
 				}
 			}
 		}
-		p.updateCache.Add(pi.Sid, updatedAt)
+		if isValid {
+			p.updateCache.Add(pi.Sid, pi.Version)
+			validUpdates = append(validUpdates, pi)
+		}
 	}
+	p.updateLock.Unlock()
+
+	if len(validUpdates) == 0 {
+		return nil
+	}
+
 	return p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Update{
 			Update: &livekit.ParticipantUpdate{
-				Participants: participantsToUpdate,
+				Participants: validUpdates,
 			},
 		},
 	})

pkg/rtc/participant_internal_test.go

@@ -172,18 +172,17 @@ func TestOutOfOrderUpdates(t *testing.T) {
 		Sid:      "PA_test2",
 		Identity: "test2",
 		Metadata: "123",
+		Version:  2,
 	}
-	earlierTs := time.Now()
-	time.Sleep(time.Millisecond)
-	laterTs := time.Now()
-	require.NoError(t, p.SendParticipantUpdate([]*livekit.ParticipantInfo{pi}, laterTs))
+	require.NoError(t, p.SendParticipantUpdate([]*livekit.ParticipantInfo{pi}))
 
 	pi = &livekit.ParticipantInfo{
 		Sid:      "PA_test2",
 		Identity: "test2",
 		Metadata: "456",
+		Version:  1,
 	}
-	require.NoError(t, p.SendParticipantUpdate([]*livekit.ParticipantInfo{pi}, earlierTs))
+	require.NoError(t, p.SendParticipantUpdate([]*livekit.ParticipantInfo{pi}))
 
 	// only sent once, and it's the earlier message
 	require.Equal(t, 1, sink.WriteMessageCallCount())

pkg/rtc/room.go

@@ -281,7 +281,7 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, responseSink routing.
 	p.SetResponseSink(responseSink)
 
 	updates := ToProtoParticipants(r.GetParticipants())
-	if err := p.SendParticipantUpdate(updates, time.Now()); err != nil {
+	if err := p.SendParticipantUpdate(updates); err != nil {
 		return err
 	}
 
@@ -699,19 +699,12 @@ func (r *Room) subscribeToExistingTracks(p types.LocalParticipant) int {
 
 // broadcast an update about participant p
 func (r *Room) broadcastParticipantState(p types.LocalParticipant, skipSource bool) {
-	//
-	// This is a critical section to ensure that participant update time and
-	// the corresponding data are paired properly.
-	//
-	r.lock.Lock()
-	updatedAt := time.Now()
 	updates := ToProtoParticipants([]types.LocalParticipant{p})
-	r.lock.Unlock()
 
 	if p.Hidden() {
 		if !skipSource {
 			// send update only to hidden participant
-			err := p.SendParticipantUpdate(updates, updatedAt)
+			err := p.SendParticipantUpdate(updates)
 			if err != nil {
 				r.Logger.Errorw("could not send update to participant", err,
 					"participant", p.Identity(), "pID", p.ID())
@@ -727,7 +720,7 @@ func (r *Room) broadcastParticipantState(p types.LocalParticipant, skipSource bo
 			continue
 		}
 
-		err := op.SendParticipantUpdate(updates, updatedAt)
+		err := op.SendParticipantUpdate(updates)
 		if err != nil {
 			r.Logger.Errorw("could not send update to participant", err,
 				"participant", p.Identity(), "pID", p.ID())

pkg/rtc/types/interfaces.go

@@ -126,7 +126,7 @@ type LocalParticipant interface {
 
 	// server sent messages
 	SendJoinResponse(info *livekit.Room, otherParticipants []*livekit.ParticipantInfo, iceServers []*livekit.ICEServer) error
-	SendParticipantUpdate(participants []*livekit.ParticipantInfo, updatedAt time.Time) error
+	SendParticipantUpdate(participants []*livekit.ParticipantInfo) error
 	SendSpeakerUpdate(speakers []*livekit.SpeakerInfo) error
 	SendDataPacket(packet *livekit.DataPacket) error
 	SendRoomUpdate(room *livekit.Room) error

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -379,11 +379,10 @@ type FakeLocalParticipant struct {
 	sendJoinResponseReturnsOnCall map[int]struct {
 		result1 error
 	}
-	SendParticipantUpdateStub        func([]*livekit.ParticipantInfo, time.Time) error
+	SendParticipantUpdateStub        func([]*livekit.ParticipantInfo) error
 	sendParticipantUpdateMutex       sync.RWMutex
 	sendParticipantUpdateArgsForCall []struct {
 		arg1 []*livekit.ParticipantInfo
-		arg2 time.Time
 	}
 	sendParticipantUpdateReturns struct {
 		result1 error
@@ -2582,7 +2581,7 @@ func (fake *FakeLocalParticipant) SendJoinResponseReturnsOnCall(i int, result1 e
 	}{result1}
 }
 
-func (fake *FakeLocalParticipant) SendParticipantUpdate(arg1 []*livekit.ParticipantInfo, arg2 time.Time) error {
+func (fake *FakeLocalParticipant) SendParticipantUpdate(arg1 []*livekit.ParticipantInfo) error {
 	var arg1Copy []*livekit.ParticipantInfo
 	if arg1 != nil {
 		arg1Copy = make([]*livekit.ParticipantInfo, len(arg1))
@@ -2592,14 +2591,13 @@ func (fake *FakeLocalParticipant) SendParticipantUpdate(arg1 []*livekit.Particip
 	ret, specificReturn := fake.sendParticipantUpdateReturnsOnCall[len(fake.sendParticipantUpdateArgsForCall)]
 	fake.sendParticipantUpdateArgsForCall = append(fake.sendParticipantUpdateArgsForCall, struct {
 		arg1 []*livekit.ParticipantInfo
-		arg2 time.Time
-	}{arg1Copy, arg2})
+	}{arg1Copy})
 	stub := fake.SendParticipantUpdateStub
 	fakeReturns := fake.sendParticipantUpdateReturns
-	fake.recordInvocation("SendParticipantUpdate", []interface{}{arg1Copy, arg2})
+	fake.recordInvocation("SendParticipantUpdate", []interface{}{arg1Copy})
 	fake.sendParticipantUpdateMutex.Unlock()
 	if stub != nil {
-		return stub(arg1, arg2)
+		return stub(arg1)
 	}
 	if specificReturn {
 		return ret.result1
@@ -2613,17 +2611,17 @@ func (fake *FakeLocalParticipant) SendParticipantUpdateCallCount() int {
 	return len(fake.sendParticipantUpdateArgsForCall)
 }
 
-func (fake *FakeLocalParticipant) SendParticipantUpdateCalls(stub func([]*livekit.ParticipantInfo, time.Time) error) {
+func (fake *FakeLocalParticipant) SendParticipantUpdateCalls(stub func([]*livekit.ParticipantInfo) error) {
 	fake.sendParticipantUpdateMutex.Lock()
 	defer fake.sendParticipantUpdateMutex.Unlock()
 	fake.SendParticipantUpdateStub = stub
 }
 
-func (fake *FakeLocalParticipant) SendParticipantUpdateArgsForCall(i int) ([]*livekit.ParticipantInfo, time.Time) {
+func (fake *FakeLocalParticipant) SendParticipantUpdateArgsForCall(i int) []*livekit.ParticipantInfo {
 	fake.sendParticipantUpdateMutex.RLock()
 	defer fake.sendParticipantUpdateMutex.RUnlock()
 	argsForCall := fake.sendParticipantUpdateArgsForCall[i]
-	return argsForCall.arg1, argsForCall.arg2
+	return argsForCall.arg1
 }
 
 func (fake *FakeLocalParticipant) SendParticipantUpdateReturns(result1 error) {

pkg/service/roommanager.go

@@ -254,10 +254,6 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam
 		return
 	}
 
-	if pi.Metadata != "" {
-		participant.SetMetadata(pi.Metadata)
-	}
-
 	// join room
 	opts := rtc.ParticipantOptions{
 		AutoSubscribe: pi.AutoSubscribe,