From 037ae572d99f0d667bd8dde6a7a47b942d2ff60c Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Fri, 3 Feb 2023 00:30:11 +0530
Subject: [PATCH] Ensure older participant session update does not go out after
 a newer (#1372)

* Ensure older participant session update does not go out after a newer
session has joined.

* fix tests

* change comment

* do not send older version
---
 pkg/rtc/room.go      | 32 ++++++++++++++++++++++----------
 pkg/rtc/room_test.go | 18 +++++++++++++++++-
 2 files changed, 39 insertions(+), 11 deletions(-)

diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go
index d5ba660f8..78cefa9fc 100644
--- a/pkg/rtc/room.go
+++ b/pkg/rtc/room.go
@@ -900,6 +900,10 @@ func (r *Room) broadcastParticipantState(p types.LocalParticipant, opts broadcas
 }
 
 func (r *Room) sendParticipantUpdates(updates []*livekit.ParticipantInfo) {
+	if len(updates) == 0 {
+		return
+	}
+
 	for _, op := range r.GetParticipants() {
 		err := op.SendParticipantUpdate(updates)
 		if err != nil {
@@ -960,16 +964,24 @@ func (r *Room) pushAndDequeueUpdates(pi *livekit.ParticipantInfo, isImmediate bo
 	shouldSend := isImmediate || pi.IsPublisher
 
 	if existing != nil {
-		if pi.Sid != existing.Sid {
-			// session change, need to send immediately
-			isImmediate = true
-			existing.State = livekit.ParticipantInfo_DISCONNECTED
-			updates = append(updates, existing)
-		} else if pi.Version < existing.Version {
-			// out of order update
-			return nil
-		} else if shouldSend {
-			updates = append(updates, existing)
+		if pi.Sid == existing.Sid {
+			// same participant session
+			if pi.Version < existing.Version {
+				// out of order update
+				return nil
+			}
+		} else {
+			// different participant sessions
+			if existing.JoinedAt < pi.JoinedAt {
+				// existing is older, synthesize a DISCONNECT for older and
+				// send immediately along with newer session to signal switch
+				shouldSend = true
+				existing.State = livekit.ParticipantInfo_DISCONNECTED
+				updates = append(updates, existing)
+			} else {
+				// older session update, newer session has already become active, so nothing to do
+				return nil
+			}
 		}
 	}
 
diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go
index c7aaf5aff..59fd9cb9c 100644
--- a/pkg/rtc/room_test.go
+++ b/pkg/rtc/room_test.go
@@ -211,28 +211,33 @@ func TestPushAndDequeueUpdates(t *testing.T) {
 		Sid:         "1",
 		IsPublisher: true,
 		Version:     1,
+		JoinedAt:    0,
 	}
 	publisher1v2 := &livekit.ParticipantInfo{
 		Identity:    identity,
 		Sid:         "1",
 		IsPublisher: true,
 		Version:     2,
+		JoinedAt:    1,
 	}
 	publisher2 := &livekit.ParticipantInfo{
 		Identity:    identity,
 		Sid:         "2",
 		IsPublisher: true,
 		Version:     1,
+		JoinedAt:    2,
 	}
 	subscriber1v1 := &livekit.ParticipantInfo{
 		Identity: identity,
 		Sid:      "1",
 		Version:  1,
+		JoinedAt: 0,
 	}
 	subscriber1v2 := &livekit.ParticipantInfo{
 		Identity: identity,
 		Sid:      "1",
 		Version:  2,
+		JoinedAt: 1,
 	}
 
 	requirePIEquals := func(t *testing.T, a, b *livekit.ParticipantInfo) {
@@ -267,6 +272,17 @@ func TestPushAndDequeueUpdates(t *testing.T) {
 				requirePIEquals(t, subscriber1v2, queued)
 			},
 		},
+		{
+			name:      "both versions updates when immediate",
+			pi:        subscriber1v2,
+			existing:  subscriber1v1,
+			immediate: true,
+			expected:  []*livekit.ParticipantInfo{subscriber1v2},
+			validate: func(t *testing.T, rm *Room, _ []*livekit.ParticipantInfo) {
+				queued := rm.batchedUpdates[livekit.ParticipantIdentity(identity)]
+				require.Nil(t, queued)
+			},
+		},
 		{
 			name:     "out of order updates are rejected",
 			pi:       subscriber1v1,
@@ -294,7 +310,7 @@ func TestPushAndDequeueUpdates(t *testing.T) {
 			name:     "when switching to publisher, queue is cleared",
 			pi:       publisher1v2,
 			existing: subscriber1v1,
-			expected: []*livekit.ParticipantInfo{subscriber1v1, publisher1v2},
+			expected: []*livekit.ParticipantInfo{publisher1v2},
 			validate: func(t *testing.T, rm *Room, updates []*livekit.ParticipantInfo) {
 				require.Empty(t, rm.batchedUpdates)
 			},