Keep track of pending reconciles to avoid duplicate queueReconcile (#1474)

pkg/rtc/participant.go

@@ -1059,7 +1059,7 @@ func (p *ParticipantImpl) setupSubscriptionManager() {
 		Telemetry:           p.params.Telemetry,
 		OnTrackSubscribed:   p.onTrackSubscribed,
 		OnTrackUnsubscribed: p.onTrackUnsubscribed,
-		OnSubcriptionError:  p.onSubscriptionError,
+		OnSubscriptionError: p.onSubscriptionError,
 	})
 }
 

pkg/rtc/subscriptionmanager.go

@@ -48,7 +48,7 @@ type SubscriptionManagerParams struct {
 	TrackResolver       types.MediaTrackResolver
 	OnTrackSubscribed   func(subTrack types.SubscribedTrack)
 	OnTrackUnsubscribed func(subTrack types.SubscribedTrack)
-	OnSubcriptionError  func(trackID livekit.TrackID)
+	OnSubscriptionError func(trackID livekit.TrackID)
 	Telemetry           telemetry.TelemetryService
 }
 
@@ -58,21 +58,24 @@ type SubscriptionManager struct {
 	lock          sync.RWMutex
 	subscriptions map[livekit.TrackID]*trackSubscription
 	subscribedTo  map[livekit.ParticipantID]map[livekit.TrackID]struct{}
-	reconcileCh   chan livekit.TrackID
-	closeCh       chan struct{}
-	doneCh        chan struct{}
+	// keeps track of tracks that are already queued for reconcile to avoid duplicating reconcile requests
+	pendingReconcile map[livekit.TrackID]struct{}
+	reconcileCh      chan livekit.TrackID
+	closeCh          chan struct{}
+	doneCh           chan struct{}
 
 	onSubscribeStatusChanged func(publisherID livekit.ParticipantID, subscribed bool)
 }
 
 func NewSubscriptionManager(params SubscriptionManagerParams) *SubscriptionManager {
 	m := &SubscriptionManager{
-		params:        params,
-		subscriptions: make(map[livekit.TrackID]*trackSubscription),
-		subscribedTo:  make(map[livekit.ParticipantID]map[livekit.TrackID]struct{}),
-		reconcileCh:   make(chan livekit.TrackID, 50),
-		closeCh:       make(chan struct{}),
-		doneCh:        make(chan struct{}),
+		params:           params,
+		subscriptions:    make(map[livekit.TrackID]*trackSubscription),
+		subscribedTo:     make(map[livekit.ParticipantID]map[livekit.TrackID]struct{}),
+		pendingReconcile: make(map[livekit.TrackID]struct{}),
+		reconcileCh:      make(chan livekit.TrackID, 50),
+		closeCh:          make(chan struct{}),
+		doneCh:           make(chan struct{}),
 	}
 
 	go m.reconcileWorker()
@@ -307,7 +310,7 @@ func (m *SubscriptionManager) reconcileSubscription(s *trackSubscription) {
 						"attempt", numAttempts,
 					)
 					s.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), err, false)
-					m.params.OnSubcriptionError(s.trackID)
+					m.params.OnSubscriptionError(s.trackID)
 				} else {
 					s.logger.Debugw("failed to subscribe, retrying",
 						"error", err,
@@ -343,13 +346,20 @@ func (m *SubscriptionManager) reconcileSubscription(s *trackSubscription) {
 		if s.durationSinceStart() > subscriptionTimeout {
 			s.logger.Errorw("track not bound after timeout", nil)
 			s.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), ErrTrackNotBound, false)
-			m.params.OnSubcriptionError(s.trackID)
+			m.params.OnSubscriptionError(s.trackID)
 		}
 	}
 }
 
-// trigger an immediate reconcilation, when trackID is empty, will reconcile all subscriptions
+// trigger an immediate reconciliation, when trackID is empty, will reconcile all subscriptions
 func (m *SubscriptionManager) queueReconcile(trackID livekit.TrackID) {
+	m.lock.Lock()
+	if _, ok := m.pendingReconcile[trackID]; ok {
+		// already reconciled
+		m.lock.Unlock()
+		return
+	}
+	m.lock.Unlock()
 	select {
 	case m.reconcileCh <- trackID:
 	default:
@@ -369,9 +379,10 @@ func (m *SubscriptionManager) reconcileWorker() {
 		case <-reconcileTicker.C:
 			m.reconcileSubscriptions()
 		case trackID := <-m.reconcileCh:
-			m.lock.RLock()
+			m.lock.Lock()
 			s := m.subscriptions[trackID]
-			m.lock.RUnlock()
+			delete(m.pendingReconcile, trackID)
+			m.lock.Unlock()
 			if s != nil {
 				m.reconcileSubscription(s)
 			} else {

pkg/rtc/subscriptionmanager_test.go

@@ -54,7 +54,7 @@ func TestSubscribe(t *testing.T) {
 		sm.params.OnTrackSubscribed = func(subTrack types.SubscribedTrack) {
 			subCount.Add(1)
 		}
-		sm.params.OnSubcriptionError = func(trackID livekit.TrackID) {
+		sm.params.OnSubscriptionError = func(trackID livekit.TrackID) {
 			failed.Store(true)
 		}
 		numParticipantSubscribed := atomic.Int32{}
@@ -115,7 +115,7 @@ func TestSubscribe(t *testing.T) {
 		resolver := newTestResolver(false, true, "pub", "pubID")
 		sm.params.TrackResolver = resolver.Resolve
 		failed := atomic.Bool{}
-		sm.params.OnSubcriptionError = func(trackID livekit.TrackID) {
+		sm.params.OnSubscriptionError = func(trackID livekit.TrackID) {
 			failed.Store(true)
 		}
 
@@ -156,7 +156,7 @@ func TestSubscribe(t *testing.T) {
 		resolver := newTestResolver(true, true, "pub", "pubID")
 		sm.params.TrackResolver = resolver.Resolve
 		failed := atomic.Bool{}
-		sm.params.OnSubcriptionError = func(trackID livekit.TrackID) {
+		sm.params.OnSubscriptionError = func(trackID livekit.TrackID) {
 			failed.Store(true)
 		}
 
@@ -278,6 +278,7 @@ func TestSubscribeStatusChanged(t *testing.T) {
 
 	require.Equal(t, int32(1), numParticipantSubscribed.Load())
 	require.Equal(t, int32(0), numParticipantUnsubscribed.Load())
+	require.True(t, sm.IsSubscribedTo("pubID"))
 
 	// now unsubscribe track2, no event should be fired
 	sm.UnsubscribeFromTrack("track2")
@@ -292,6 +293,7 @@ func TestSubscribeStatusChanged(t *testing.T) {
 		return !s1.needsUnsubscribe()
 	}, subSettleTimeout, subCheckInterval, "track1 should be unsubscribed")
 	require.Equal(t, int32(1), numParticipantUnsubscribed.Load())
+	require.False(t, sm.IsSubscribedTo("pubID"))
 }
 
 // clients may send update subscribed settings prior to subscription events coming through
@@ -334,7 +336,7 @@ func newTestSubscriptionManager(t *testing.T) *SubscriptionManager {
 		Logger:              logger.GetLogger(),
 		OnTrackSubscribed:   func(subTrack types.SubscribedTrack) {},
 		OnTrackUnsubscribed: func(subTrack types.SubscribedTrack) {},
-		OnSubcriptionError:  func(trackID livekit.TrackID) {},
+		OnSubscriptionError: func(trackID livekit.TrackID) {},
 		TrackResolver: func(identity livekit.ParticipantIdentity, trackID livekit.TrackID) types.MediaResolverResult {
 			return types.MediaResolverResult{}
 		},