From 94564f9f93686c20d37c68a4e0298eb11f60c91b Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Sat, 24 Dec 2022 12:53:18 +0530
Subject: [PATCH] Introduce a safety note for dupe detection. (#1257)

If the last non-dupe was more than a while ago, just let one through
even if it is dupe.

Clean up code.
---
 pkg/rtc/signaldeduper/subscriptiondeduper.go | 83 ++++++++++++--------
 1 file changed, 51 insertions(+), 32 deletions(-)

diff --git a/pkg/rtc/signaldeduper/subscriptiondeduper.go b/pkg/rtc/signaldeduper/subscriptiondeduper.go
index 6c85a10bc..c8622a1f9 100644
--- a/pkg/rtc/signaldeduper/subscriptiondeduper.go
+++ b/pkg/rtc/signaldeduper/subscriptiondeduper.go
@@ -2,6 +2,7 @@ package signaldeduper
 
 import (
 	"sync"
+	"time"
 
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
@@ -9,6 +10,10 @@ import (
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 )
 
+const (
+	dupeBarrierDuration = 5 * time.Second
+)
+
 // --------------------------------------------------
 
 type subscriptionSetting struct {
@@ -45,17 +50,22 @@ func (s *subscriptionSetting) Equal(other *subscriptionSetting) bool {
 
 // --------------------------------------------------
 
+type subscriptionState struct {
+	setting         *subscriptionSetting
+	lastNonDupeTime time.Time
+}
+
 type SubscriptionDeduper struct {
 	logger logger.Logger
 
 	lock                      sync.RWMutex
-	participantsSubscriptions map[livekit.ParticipantKey]map[livekit.TrackID]*subscriptionSetting
+	participantsSubscriptions map[livekit.ParticipantKey]map[livekit.TrackID]*subscriptionState
 }
 
 func NewSubscriptionDeduper(logger logger.Logger) types.SignalDeduper {
 	return &SubscriptionDeduper{
 		logger:                    logger,
-		participantsSubscriptions: make(map[livekit.ParticipantKey]map[livekit.TrackID]*subscriptionSetting),
+		participantsSubscriptions: make(map[livekit.ParticipantKey]map[livekit.TrackID]*subscriptionState),
 	}
 }
 
@@ -87,8 +97,6 @@ func (s *SubscriptionDeduper) updateSubscriptionsFromUpdateSubscription(
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	participantSubscriptions := s.getOrCreateParticipantSubscriptions(participantKey)
-
 	numTracks := len(us.TrackSids)
 	for _, pt := range us.ParticipantTracks {
 		numTracks += len(pt.TrackSids)
@@ -103,20 +111,11 @@ func (s *SubscriptionDeduper) updateSubscriptionsFromUpdateSubscription(
 		}
 	}
 
+	newSetting := subscriptionSettingFromUpdateSubscription(us)
 	for trackID := range trackIDs {
-		currentSetting := participantSubscriptions[trackID]
-		if currentSetting == nil {
-			// new track seen
-			currentSetting = subscriptionSettingFromUpdateSubscription(us)
-			participantSubscriptions[trackID] = currentSetting
+		isTrackDupe := s.detectDupe(participantKey, trackID, newSetting)
+		if !isTrackDupe {
 			isDupe = false
-		} else {
-			newSetting := subscriptionSettingFromUpdateSubscription(us)
-			if !currentSetting.Equal(newSetting) {
-				// subscription setting change
-				participantSubscriptions[trackID] = newSetting
-				isDupe = false
-			}
 		}
 	}
 
@@ -132,34 +131,54 @@ func (s *SubscriptionDeduper) updateSubscriptionsFromUpdateTrackSettings(
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	participantSubscriptions := s.getOrCreateParticipantSubscriptions(participantKey)
-
+	newSetting := subscriptionSettingFromUpdateTrackSettings(uts)
 	for _, trackSid := range uts.TrackSids {
-		currentSetting := participantSubscriptions[livekit.TrackID(trackSid)]
-		if currentSetting == nil {
-			// new track seen
-			currentSetting = subscriptionSettingFromUpdateTrackSettings(uts)
-			participantSubscriptions[livekit.TrackID(trackSid)] = currentSetting
+		isTrackDupe := s.detectDupe(participantKey, livekit.TrackID(trackSid), newSetting)
+		if !isTrackDupe {
 			isDupe = false
-		} else {
-			newSetting := subscriptionSettingFromUpdateTrackSettings(uts)
-			if !currentSetting.Equal(newSetting) {
-				// subscription setting change
-				participantSubscriptions[livekit.TrackID(trackSid)] = newSetting
-				isDupe = false
-			}
 		}
 	}
 
 	return isDupe
 }
 
-func (s *SubscriptionDeduper) getOrCreateParticipantSubscriptions(participantKey livekit.ParticipantKey) map[livekit.TrackID]*subscriptionSetting {
+func (s *SubscriptionDeduper) getOrCreateParticipantSubscriptions(
+	participantKey livekit.ParticipantKey,
+) map[livekit.TrackID]*subscriptionState {
 	participantSubscriptions := s.participantsSubscriptions[participantKey]
 	if participantSubscriptions == nil {
-		participantSubscriptions = make(map[livekit.TrackID]*subscriptionSetting)
+		participantSubscriptions = make(map[livekit.TrackID]*subscriptionState)
 		s.participantsSubscriptions[participantKey] = participantSubscriptions
 	}
 
 	return participantSubscriptions
 }
+
+func (s *SubscriptionDeduper) detectDupe(
+	participantKey livekit.ParticipantKey,
+	trackID livekit.TrackID,
+	updatedSetting *subscriptionSetting,
+) bool {
+	isDupe := true
+	participantSubscriptions := s.getOrCreateParticipantSubscriptions(participantKey)
+	state := participantSubscriptions[trackID]
+	if state == nil || !state.setting.Equal(updatedSetting) {
+		// new track seen or subscription setting change
+		state = &subscriptionState{
+			setting:         updatedSetting,
+			lastNonDupeTime: time.Now(),
+		}
+		isDupe = false
+	}
+
+	if isDupe && time.Since(state.lastNonDupeTime) > dupeBarrierDuration {
+		state.lastNonDupeTime = time.Now()
+		isDupe = false
+	}
+
+	if !isDupe {
+		participantSubscriptions[trackID] = state
+	}
+
+	return isDupe
+}