livekit/pkg/rtc/subscriptionmanager_test.go

/*
 * Copyright 2023 LiveKit, Inc
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package rtc

import (
	"fmt"
	"sync"
	"testing"
	"time"

	"github.com/pion/webrtc/v4"
	"github.com/stretchr/testify/require"
	"go.uber.org/atomic"

	"github.com/livekit/livekit-server/pkg/rtc/types"
	"github.com/livekit/livekit-server/pkg/rtc/types/typesfakes"
	"github.com/livekit/livekit-server/pkg/utils"
	"github.com/livekit/protocol/livekit"
	"github.com/livekit/protocol/logger"
)

func init() {
	reconcileInterval = 50 * time.Millisecond
	notFoundTimeout = 200 * time.Millisecond
	subscriptionTimeout = 200 * time.Millisecond
}

const (
	subSettleTimeout = 600 * time.Millisecond
	subCheckInterval = 10 * time.Millisecond
)

func TestSubscribe(t *testing.T) {
	t.Run("happy path subscribe", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		resolver := newTestResolver(true, true, "pub", "pubID")
		sm.params.TrackResolver = resolver.Resolve
		subCount := atomic.Int32{}
		failed := atomic.Bool{}
		sm.params.OnTrackSubscribed = func(subTrack types.SubscribedTrack) {
			subCount.Add(1)
		}
		sm.params.OnSubscriptionError = func(trackID livekit.TrackID, fatal bool, err error) {
			failed.Store(true)
		}
		numParticipantSubscribed := atomic.Int32{}
		numParticipantUnsubscribed := atomic.Int32{}
		sm.OnSubscribeStatusChanged(func(pubID livekit.ParticipantID, subscribed bool) {
			if subscribed {
				numParticipantSubscribed.Add(1)
			} else {
				numParticipantUnsubscribed.Add(1)
			}
		})

		sm.SubscribeToTrack("track", false)
		s := sm.subscriptions["track"]
		require.True(t, s.isDesired())
		require.Eventually(t, func() bool {
			return subCount.Load() == 1
		}, subSettleTimeout, subCheckInterval, "track was not subscribed")

		require.NotNil(t, s.getSubscribedTrack())
		require.Len(t, sm.GetSubscribedTracks(), 1)

		require.Eventually(t, func() bool {
			return len(sm.GetSubscribedParticipants()) == 1
		}, subSettleTimeout, subCheckInterval, "GetSubscribedParticipants should have returned one item")
		require.Equal(t, "pubID", string(sm.GetSubscribedParticipants()[0]))

		// ensure telemetry events are sent
		tl := sm.params.Participant.GetTelemetryListener().(*typesfakes.FakeParticipantTelemetryListener)
		require.Equal(t, 1, tl.OnTrackSubscribeRequestedCallCount())

		// ensure bound
		setTestSubscribedTrackBound(t, s.getSubscribedTrack())
		require.Eventually(t, func() bool {
			return !s.needsBind()
		}, subSettleTimeout, subCheckInterval, "track was not bound")

		// telemetry event should have been sent
		require.Equal(t, 1, tl.OnTrackSubscribedCallCount())

		time.Sleep(notFoundTimeout)
		require.False(t, failed.Load())

		resolver.SetPause(true)
		// ensure its resilience after being closed
		setTestSubscribedTrackClosed(t, s.getSubscribedTrack(), false)
		require.Eventually(t, func() bool {
			return s.needsSubscribe()
		}, subSettleTimeout, subCheckInterval, "needs subscribe did not persist across track close")
		resolver.SetPause(false)

		require.Eventually(t, func() bool {
			return s.isDesired() && !s.needsSubscribe()
		}, subSettleTimeout, subCheckInterval, "track was not resubscribed")

		// was subscribed twice, unsubscribed once (due to close)
		require.Eventually(t, func() bool {
			return numParticipantSubscribed.Load() == 2
		}, subSettleTimeout, subCheckInterval, "participant subscribe status was not updated twice")
		require.Eventually(t, func() bool {
			return numParticipantUnsubscribed.Load() == 1
		}, subSettleTimeout, subCheckInterval, "participant unsubscribe status was not updated")
	})

	t.Run("no track permission", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		resolver := newTestResolver(false, true, "pub", "pubID")
		sm.params.TrackResolver = resolver.Resolve
		failed := atomic.Bool{}
		sm.params.OnSubscriptionError = func(trackID livekit.TrackID, fatal bool, err error) {
			failed.Store(true)
		}

		sm.SubscribeToTrack("track", false)
		s := sm.subscriptions["track"]
		require.Eventually(t, func() bool {
			return !s.getHasPermission()
		}, subSettleTimeout, subCheckInterval, "should not have permission to subscribe")

		time.Sleep(subscriptionTimeout)

		// should not have called failed callbacks, isDesired remains unchanged
		require.True(t, s.isDesired())
		require.False(t, failed.Load())
		require.True(t, s.needsSubscribe())
		require.Len(t, sm.GetSubscribedTracks(), 0)

		// trackSubscribed telemetry not sent
		tl := sm.params.Participant.GetTelemetryListener().(*typesfakes.FakeParticipantTelemetryListener)
		require.Equal(t, 1, tl.OnTrackSubscribeRequestedCallCount())
		require.Equal(t, 0, tl.OnTrackSubscribedCallCount())

		// give permissions now
		resolver.lock.Lock()
		resolver.hasPermission = true
		resolver.lock.Unlock()

		require.Eventually(t, func() bool {
			return !s.needsSubscribe()
		}, subSettleTimeout, subCheckInterval, "should be subscribed")

		require.Len(t, sm.GetSubscribedTracks(), 1)
	})

	t.Run("publisher left", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		resolver := newTestResolver(true, true, "pub", "pubID")
		sm.params.TrackResolver = resolver.Resolve
		failed := atomic.Bool{}
		sm.params.OnSubscriptionError = func(trackID livekit.TrackID, fatal bool, err error) {
			failed.Store(true)
		}

		sm.SubscribeToTrack("track", false)
		s := sm.subscriptions["track"]
		require.Eventually(t, func() bool {
			return !s.needsSubscribe()
		}, subSettleTimeout, subCheckInterval, "should be subscribed")

		resolver.lock.Lock()
		resolver.hasTrack = false
		resolver.lock.Unlock()

		// publisher triggers close
		setTestSubscribedTrackClosed(t, s.getSubscribedTrack(), false)

		require.Eventually(t, func() bool {
			return !s.isDesired()
		}, subSettleTimeout, subCheckInterval, "isDesired not set to false")
	})
}

func TestUnsubscribe(t *testing.T) {
	sm := newTestSubscriptionManager()
	defer sm.Close(false)
	unsubCount := atomic.Int32{}
	sm.params.OnTrackUnsubscribed = func(subTrack types.SubscribedTrack) {
		unsubCount.Add(1)
	}

	resolver := newTestResolver(true, true, "pub", "pubID")

	s := &mediaTrackSubscription{
		trackSubscription: trackSubscription{
			trackID:           "track",
			desired:           true,
			subscriberID:      sm.params.Participant.ID(),
			publisherID:       "pubID",
			publisherIdentity: "pub",
			logger:            logger.GetLogger(),
			hasPermission:     true,
		},
		bound: true,
	}
	// a bunch of unfortunate manual wiring
	res := resolver.Resolve(nil, s.trackID)
	res.TrackChangedNotifier.AddObserver(string(sm.params.Participant.ID()), func() {})
	s.changedNotifier = res.TrackChangedNotifier
	st, err := res.Track.AddSubscriber(sm.params.Participant)
	require.NoError(t, err)
	s.subscribedTrack = st
	st.OnClose(func(isExpectedToResume bool) {
		sm.handleSubscribedTrackClose(s, isExpectedToResume)
	})
	res.Track.(*typesfakes.FakeMediaTrack).RemoveSubscriberCalls(func(pID livekit.ParticipantID, isExpectedToResume bool) {
		setTestSubscribedTrackClosed(t, st, isExpectedToResume)
	})

	sm.lock.Lock()
	sm.subscriptions["track"] = s
	sm.lock.Unlock()

	require.False(t, s.needsSubscribe())
	require.False(t, s.needsUnsubscribe())

	// unsubscribe
	sm.UnsubscribeFromTrack("track")
	require.False(t, s.isDesired())

	require.Eventually(t, func() bool {
		if s.needsUnsubscribe() {
			return false
		}
		if sm.pendingUnsubscribes.Load() != 0 {
			return false
		}
		sm.lock.RLock()
		subLen := len(sm.subscriptions)
		sm.lock.RUnlock()
		return subLen == 0
	}, subSettleTimeout, subCheckInterval, "Track was not unsubscribed")

	// no traces should be left
	require.Len(t, sm.GetSubscribedTracks(), 0)
	// the observer is dropped on a goroutine of its own
	require.Eventually(t, func() bool {
		return !res.TrackChangedNotifier.HasObservers()
	}, subSettleTimeout, subCheckInterval, "observer was not removed")

	tl := sm.params.Participant.GetTelemetryListener().(*typesfakes.FakeParticipantTelemetryListener)
	require.Equal(t, 1, tl.OnTrackUnsubscribedCallCount())
}

func TestSubscribeStatusChanged(t *testing.T) {
	sm := newTestSubscriptionManager()
	defer sm.Close(false)
	resolver := newTestResolver(true, true, "pub", "pubID")
	sm.params.TrackResolver = resolver.Resolve
	numParticipantSubscribed := atomic.Int32{}
	numParticipantUnsubscribed := atomic.Int32{}
	sm.OnSubscribeStatusChanged(func(pubID livekit.ParticipantID, subscribed bool) {
		if subscribed {
			numParticipantSubscribed.Add(1)
		} else {
			numParticipantUnsubscribed.Add(1)
		}
	})

	sm.SubscribeToTrack("track1", false)
	sm.SubscribeToTrack("track2", false)
	s1 := sm.subscriptions["track1"]
	s2 := sm.subscriptions["track2"]
	require.Eventually(t, func() bool {
		return !s1.needsSubscribe() && !s2.needsSubscribe()
	}, subSettleTimeout, subCheckInterval, "track1 and track2 should be subscribed")
	st1 := s1.getSubscribedTrack()
	st1.OnClose(func(isExpectedToResume bool) {
		sm.handleSubscribedTrackClose(s1, isExpectedToResume)
	})
	st2 := s2.getSubscribedTrack()
	st2.OnClose(func(isExpectedToResume bool) {
		sm.handleSubscribedTrackClose(s2, isExpectedToResume)
	})
	st1.MediaTrack().(*typesfakes.FakeMediaTrack).RemoveSubscriberCalls(func(pID livekit.ParticipantID, isExpectedToResume bool) {
		setTestSubscribedTrackClosed(t, st1, isExpectedToResume)
	})
	st2.MediaTrack().(*typesfakes.FakeMediaTrack).RemoveSubscriberCalls(func(pID livekit.ParticipantID, isExpectedToResume bool) {
		setTestSubscribedTrackClosed(t, st2, isExpectedToResume)
	})

	require.Eventually(t, func() bool {
		return numParticipantSubscribed.Load() == 1
	}, subSettleTimeout, subCheckInterval, "should be subscribed to publisher")
	require.Equal(t, int32(0), numParticipantUnsubscribed.Load())
	require.True(t, sm.IsSubscribedTo("pubID"))

	// now unsubscribe track2, no event should be fired
	sm.UnsubscribeFromTrack("track2")
	require.Eventually(t, func() bool {
		return !s2.needsUnsubscribe()
	}, subSettleTimeout, subCheckInterval, "track2 should be unsubscribed")
	require.Equal(t, int32(0), numParticipantUnsubscribed.Load())

	// unsubscribe track1, expect event
	sm.UnsubscribeFromTrack("track1")
	require.Eventually(t, func() bool {
		return !s1.needsUnsubscribe()
	}, subSettleTimeout, subCheckInterval, "track1 should be unsubscribed")
	require.Eventually(t, func() bool {
		return numParticipantUnsubscribed.Load() == 1
	}, subSettleTimeout, subCheckInterval, "should be subscribed to publisher")
	require.False(t, sm.IsSubscribedTo("pubID"))
}

// clients may send update subscribed settings prior to subscription events coming through
// settings should be persisted and used when the subscription does take place.
func TestUpdateSettingsBeforeSubscription(t *testing.T) {
	sm := newTestSubscriptionManager()
	defer sm.Close(false)
	resolver := newTestResolver(true, true, "pub", "pubID")
	sm.params.TrackResolver = resolver.Resolve

	settings := &livekit.UpdateTrackSettings{
		Disabled: true,
		Width:    100,
		Height:   100,
	}
	sm.UpdateSubscribedTrackSettings("track", settings)

	sm.SubscribeToTrack("track", false)

	s := sm.subscriptions["track"]
	require.Eventually(t, func() bool {
		return !s.needsSubscribe()
	}, subSettleTimeout, subCheckInterval, "Track should be subscribed")

	st := s.getSubscribedTrack().(*typesfakes.FakeSubscribedTrack)
	require.Eventually(t, func() bool {
		return st.UpdateSubscriberSettingsCallCount() == 1
	}, subSettleTimeout, subCheckInterval, "UpdateSubscriberSettings should be called once")

	applied, _ := st.UpdateSubscriberSettingsArgsForCall(0)
	require.Equal(t, settings.Disabled, applied.Disabled)
	require.Equal(t, settings.Width, applied.Width)
	require.Equal(t, settings.Height, applied.Height)
}

func TestConcurrentFirstSubscribe(t *testing.T) {
	settings := &livekit.UpdateTrackSettings{Width: 100, Height: 100}
	for range 200 {
		sm := newTestSubscriptionManager()

		var lock sync.Mutex
		var subscribed bool
		mt := &typesfakes.FakeMediaTrack{}
		mt.IDReturns("track")
		mt.AddSubscriberCalls(func(types.LocalParticipant) (types.SubscribedTrack, error) {
			lock.Lock()
			defer lock.Unlock()
			if subscribed {
				return nil, errAlreadySubscribed
			}
			subscribed = true
			st := &typesfakes.FakeSubscribedTrack{}
			st.IDReturns("track")
			st.MediaTrackReturns(mt)
			return st, nil
		})
		sm.params.TrackResolver = func(types.LocalParticipant, livekit.TrackID) types.MediaResolverResult {
			return types.MediaResolverResult{
				Track:                mt,
				HasPermission:        true,
				PublisherID:          "pubID",
				PublisherIdentity:    "pub",
				TrackChangedNotifier: utils.NewChangeNotifier(),
				TrackRemovedNotifier: utils.NewChangeNotifier(),
			}
		}

		// two first subscribes and a settings update race to create the subscription
		var start, done sync.WaitGroup
		start.Add(1)
		for _, f := range []func(){
			func() { sm.SubscribeToTrack("track", false) },
			func() { sm.SubscribeToTrack("track", false) },
			func() { sm.UpdateSubscribedTrackSettings("track", settings) },
		} {
			done.Add(1)
			go func() {
				defer done.Done()
				start.Wait()
				f()
			}()
		}
		start.Done()
		done.Wait()

		sm.lock.RLock()
		s := sm.subscriptions["track"]
		sm.lock.RUnlock()
		require.Eventually(t, func() bool {
			return !s.needsSubscribe()
		}, subSettleTimeout, subCheckInterval, "the subscription that is kept should own the down track")
		st := s.getSubscribedTrack().(*typesfakes.FakeSubscribedTrack)
		require.Eventually(t, func() bool {
			n := st.UpdateSubscriberSettingsCallCount()
			if n == 0 {
				return false
			}
			applied, _ := st.UpdateSubscriberSettingsArgsForCall(n - 1)
			return applied == settings
		}, subSettleTimeout, subCheckInterval, "the down track should get the settings")

		sm.Close(false)
	}
}

func TestSettingsKeptForSubscribe(t *testing.T) {
	t.Run("media", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		resolver := newTestResolver(true, true, "pub", "pubID")
		sm.params.TrackResolver = resolver.Resolve

		settings := &livekit.UpdateTrackSettings{Disabled: true}
		sm.UpdateSubscribedTrackSettings("track", settings)
		// a cleanup pass between the settings and the subscribe must not drop the settings
		sm.reconcileSubscriptions()
		sm.SubscribeToTrack("track", false)

		sm.lock.RLock()
		s := sm.subscriptions["track"]
		sm.lock.RUnlock()
		require.Eventually(t, func() bool {
			return !s.needsSubscribe()
		}, subSettleTimeout, subCheckInterval, "track should be subscribed")
		st := s.getSubscribedTrack().(*typesfakes.FakeSubscribedTrack)
		require.Eventually(t, func() bool {
			n := st.UpdateSubscriberSettingsCallCount()
			if n == 0 {
				return false
			}
			applied, _ := st.UpdateSubscriberSettingsArgsForCall(n - 1)
			return applied == settings
		}, subSettleTimeout, subCheckInterval, "the down track should get the settings")

		// settings for a track that is never subscribed are cleaned up after notFoundTimeout
		sm.UpdateSubscribedTrackSettings("other", settings)
		require.Eventually(t, func() bool {
			sm.lock.RLock()
			defer sm.lock.RUnlock()
			_, ok := sm.subscriptions["other"]
			return !ok
		}, subSettleTimeout, subCheckInterval, "unused settings should be cleaned up")
	})

	t.Run("data", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		resolver := newTestDataTrackResolver(true, true, "pub", "pubID")
		sm.params.DataTrackResolver = resolver.Resolve

		fps := uint32(5)
		options := &livekit.DataTrackSubscriptionOptions{TargetFps: &fps}
		sm.UpdateDataTrackSubscriptionOptions("track", options)
		// a cleanup pass between the options and the subscribe must not drop the options
		sm.reconcileDataTrackSubscriptions()
		sm.SubscribeToDataTrack("track")

		sm.lock.RLock()
		s := sm.dataTrackSubscriptions["track"]
		sm.lock.RUnlock()
		require.Eventually(t, func() bool {
			return s.getDataDownTrack() != nil
		}, subSettleTimeout, subCheckInterval, "data track should be subscribed")
		ddt := s.getDataDownTrack().(*typesfakes.FakeDataDownTrack)
		n := ddt.UpdateSubscriptionOptionsCallCount()
		require.NotZero(t, n)
		require.Equal(t, options, ddt.UpdateSubscriptionOptionsArgsForCall(n-1))
	})
}

func TestSubscribeBurstOverflowsQueue(t *testing.T) {
	// a dropped reconcile must not wait for the ticker
	saved := reconcileInterval
	reconcileInterval = time.Hour
	defer func() { reconcileInterval = saved }()

	t.Run("media", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)

		// the first subscribe holds the worker until all are queued, so the queue overflows
		release := make(chan struct{})
		var holdOnce sync.Once
		sm.params.TrackResolver = func(_ types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult {
			mt := &typesfakes.FakeMediaTrack{}
			mt.IDReturns(trackID)
			mt.AddSubscriberCalls(func(types.LocalParticipant) (types.SubscribedTrack, error) {
				holdOnce.Do(func() { <-release })
				st := &typesfakes.FakeSubscribedTrack{}
				st.IDReturns(trackID)
				st.MediaTrackReturns(mt)
				return st, nil
			})
			return types.MediaResolverResult{
				Track:                mt,
				HasPermission:        true,
				PublisherID:          "pubID",
				PublisherIdentity:    "pub",
				TrackChangedNotifier: utils.NewChangeNotifier(),
				TrackRemovedNotifier: utils.NewChangeNotifier(),
			}
		}

		numTracks := cap(sm.reconcileCh) + 10
		for i := range numTracks {
			sm.SubscribeToTrack(livekit.TrackID(fmt.Sprintf("TR_%d", i)), false)
		}
		close(release)

		require.Eventually(t, func() bool {
			sm.lock.RLock()
			defer sm.lock.RUnlock()
			for _, s := range sm.subscriptions {
				if s.needsSubscribe() {
					return false
				}
			}
			return len(sm.subscriptions) == numTracks
		}, subSettleTimeout, subCheckInterval, "every track should be subscribed")
	})

	t.Run("data", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		resolver := newTestDataTrackResolver(true, true, "pub", "pubID")
		sm.params.DataTrackResolver = resolver.Resolve

		release := make(chan struct{})
		var holdOnce sync.Once
		resolver.dataTrack.AddSubscriberCalls(func(types.LocalParticipant) (types.DataDownTrack, error) {
			holdOnce.Do(func() { <-release })
			ddt := &typesfakes.FakeDataDownTrack{}
			ddt.PublishDataTrackReturns(resolver.dataTrack)
			return ddt, nil
		})

		numTracks := cap(sm.reconcileDataTrackCh) + 10
		for i := range numTracks {
			sm.SubscribeToDataTrack(livekit.TrackID(fmt.Sprintf("DTR_%d", i)))
		}
		close(release)

		require.Eventually(t, func() bool {
			sm.lock.RLock()
			defer sm.lock.RUnlock()
			for _, s := range sm.dataTrackSubscriptions {
				if s.needsSubscribe() {
					return false
				}
			}
			return len(sm.dataTrackSubscriptions) == numTracks
		}, subSettleTimeout, subCheckInterval, "every data track should be subscribed")
	})
}

func TestSubscriptionLimits(t *testing.T) {
	sm := newTestSubscriptionManagerWithParams(testSubscriptionParams{
		SubscriptionLimitAudio: 1,
		SubscriptionLimitVideo: 1,
	})
	defer sm.Close(false)
	resolver := newTestResolver(true, true, "pub", "pubID")
	sm.params.TrackResolver = resolver.Resolve
	subCount := atomic.Int32{}
	failed := atomic.Bool{}
	sm.params.OnTrackSubscribed = func(subTrack types.SubscribedTrack) {
		subCount.Add(1)
	}
	sm.params.OnSubscriptionError = func(trackID livekit.TrackID, fatal bool, err error) {
		failed.Store(true)
	}
	numParticipantSubscribed := atomic.Int32{}
	numParticipantUnsubscribed := atomic.Int32{}
	sm.OnSubscribeStatusChanged(func(pubID livekit.ParticipantID, subscribed bool) {
		if subscribed {
			numParticipantSubscribed.Add(1)
		} else {
			numParticipantUnsubscribed.Add(1)
		}
	})

	sm.SubscribeToTrack("track", false)
	s := sm.subscriptions["track"]
	require.True(t, s.isDesired())
	require.Eventually(t, func() bool {
		return subCount.Load() == 1
	}, subSettleTimeout, subCheckInterval, "track was not subscribed")

	require.NotNil(t, s.getSubscribedTrack())
	require.Len(t, sm.GetSubscribedTracks(), 1)

	require.Eventually(t, func() bool {
		return len(sm.GetSubscribedParticipants()) == 1
	}, subSettleTimeout, subCheckInterval, "GetSubscribedParticipants should have returned one item")
	require.Equal(t, "pubID", string(sm.GetSubscribedParticipants()[0]))

	// ensure telemetry events are sent
	tl := sm.params.Participant.GetTelemetryListener().(*typesfakes.FakeParticipantTelemetryListener)
	require.Equal(t, 1, tl.OnTrackSubscribeRequestedCallCount())

	// ensure bound
	setTestSubscribedTrackBound(t, s.getSubscribedTrack())
	require.Eventually(t, func() bool {
		return !s.needsBind()
	}, subSettleTimeout, subCheckInterval, "track was not bound")

	// telemetry event should have been sent
	require.Equal(t, 1, tl.OnTrackSubscribedCallCount())

	// reach subscription limit, subscribe pending
	sm.SubscribeToTrack("track2", false)
	s2 := sm.subscriptions["track2"]
	time.Sleep(subscriptionTimeout * 2)
	require.True(t, s2.needsSubscribe())
	require.Equal(t, 2, tl.OnTrackSubscribeRequestedCallCount())
	require.Equal(t, 1, tl.OnTrackSubscribeFailedCallCount())
	require.Len(t, sm.GetSubscribedTracks(), 1)

	// unsubscribe track1, then track2 should be subscribed
	sm.UnsubscribeFromTrack("track")
	require.False(t, s.isDesired())
	require.True(t, s.needsUnsubscribe())
	// wait for unsubscribe to take effect
	time.Sleep(reconcileInterval)
	setTestSubscribedTrackClosed(t, s.getSubscribedTrack(), false)
	require.Nil(t, s.getSubscribedTrack())

	time.Sleep(reconcileInterval)
	require.True(t, s2.isDesired())
	require.False(t, s2.needsSubscribe())
	require.EqualValues(t, 2, subCount.Load())
	require.NotNil(t, s2.getSubscribedTrack())
	require.Equal(t, 2, tl.OnTrackSubscribeRequestedCallCount())
	require.Len(t, sm.GetSubscribedTracks(), 1)

	// ensure bound
	setTestSubscribedTrackBound(t, s2.getSubscribedTrack())
	require.Eventually(t, func() bool {
		return !s2.needsBind()
	}, subSettleTimeout, subCheckInterval, "track was not bound")

	// subscribe to track1 again, which should pending
	sm.SubscribeToTrack("track", false)
	s = sm.subscriptions["track"]
	require.True(t, s.isDesired())
	time.Sleep(subscriptionTimeout * 2)
	require.True(t, s.needsSubscribe())
	require.Equal(t, 3, tl.OnTrackSubscribeRequestedCallCount())
	require.Equal(t, 2, tl.OnTrackSubscribeFailedCallCount())
	require.Len(t, sm.GetSubscribedTracks(), 1)
}

func TestFailTrackBind(t *testing.T) {
	sm := newTestSubscriptionManager()
	defer sm.Close(false)
	resolver := newTestResolver(true, true, "pub", "pubID")
	sm.params.TrackResolver = resolver.Resolve
	var subErr atomic.Error
	sm.params.OnSubscriptionError = func(trackID livekit.TrackID, fatal bool, err error) {
		subErr.Store(err)
	}

	sm.SubscribeToTrack("track", false)
	s := sm.subscriptions["track"]
	require.Eventually(t, func() bool {
		return s.getSubscribedTrack() != nil
	}, subSettleTimeout, subCheckInterval, "track was not subscribed")

	sm.FailTrackBind("track", webrtc.ErrUnsupportedCodec)

	tl := sm.params.Participant.GetTelemetryListener().(*typesfakes.FakeParticipantTelemetryListener)
	require.Equal(t, 1, tl.OnTrackSubscribeFailedCallCount())
	_, trackID, err, isUserError := tl.OnTrackSubscribeFailedArgsForCall(0)
	require.Equal(t, livekit.TrackID("track"), trackID)
	require.ErrorIs(t, err, webrtc.ErrUnsupportedCodec)
	require.True(t, isUserError)
	require.ErrorIs(t, subErr.Load(), webrtc.ErrUnsupportedCodec)
	require.False(t, s.isDesired())

	// an unknown track is ignored
	sm.FailTrackBind("unknown", webrtc.ErrUnsupportedCodec)
	require.Equal(t, 1, tl.OnTrackSubscribeFailedCallCount())
}

func TestSubscribeDataTrack(t *testing.T) {
	t.Run("no track permission", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		resolver := newTestDataTrackResolver(false, true, "pub", "pubID")
		sm.params.DataTrackResolver = resolver.Resolve

		sm.SubscribeToDataTrack("track")
		sm.lock.RLock()
		s := sm.dataTrackSubscriptions["track"]
		sm.lock.RUnlock()
		require.Eventually(t, func() bool {
			return !s.getHasPermission()
		}, subSettleTimeout, subCheckInterval, "should not have permission to subscribe")

		time.Sleep(subscriptionTimeout)

		// isDesired remains unchanged, no subscriber added to the data track
		require.True(t, s.isDesired())
		require.True(t, s.needsSubscribe())
		require.Equal(t, 0, resolver.dataTrack.AddSubscriberCallCount())

		// subscriber should have been notified that permission was denied
		p := sm.params.Participant.(*typesfakes.FakeLocalParticipant)
		require.Equal(t, 1, p.SendSubscriptionPermissionUpdateCallCount())
		_, _, allowed := p.SendSubscriptionPermissionUpdateArgsForCall(0)
		require.False(t, allowed)

		// give permissions now
		resolver.SetHasPermission(true)

		require.Eventually(t, func() bool {
			return !s.needsSubscribe()
		}, subSettleTimeout, subCheckInterval, "should be subscribed")
		require.NotNil(t, s.getDataDownTrack())
		require.Equal(t, 1, resolver.dataTrack.AddSubscriberCallCount())
	})

	t.Run("permission revoked while subscribed", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		resolver := newTestDataTrackResolver(true, true, "pub", "pubID")
		sm.params.DataTrackResolver = resolver.Resolve

		sm.SubscribeToDataTrack("track")
		sm.lock.RLock()
		s := sm.dataTrackSubscriptions["track"]
		sm.lock.RUnlock()
		require.Eventually(t, func() bool {
			return !s.needsSubscribe()
		}, subSettleTimeout, subCheckInterval, "should be subscribed")

		// revoke permission and close the down track like the publisher would
		// (DataTrack.RevokeDisallowedSubscribers -> RemoveSubscriber -> DataDownTrack.Close)
		resolver.SetHasPermission(false)
		ddt := s.getDataDownTrack()
		require.NotNil(t, ddt)
		setTestDataDownTrackClosed(t, ddt)

		// subscription stays desired but is denied on resubscribe attempts
		require.Nil(t, s.getDataDownTrack())
		require.True(t, s.isDesired())
		require.Eventually(t, func() bool {
			return !s.getHasPermission()
		}, subSettleTimeout, subCheckInterval, "should not have permission to resubscribe")
		require.True(t, s.needsSubscribe())
		require.Equal(t, 1, resolver.dataTrack.AddSubscriberCallCount())

		// give permission back, should resubscribe
		resolver.SetHasPermission(true)

		require.Eventually(t, func() bool {
			return !s.needsSubscribe() && s.getDataDownTrack() != nil
		}, subSettleTimeout, subCheckInterval, "should be resubscribed")
		require.Equal(t, 2, resolver.dataTrack.AddSubscriberCallCount())
	})

	t.Run("subscribe again during unsubscribe", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		resolver := newTestDataTrackResolver(true, true, "pub", "pubID")
		sm.params.DataTrackResolver = resolver.Resolve

		sm.SubscribeToDataTrack("track")
		sm.lock.RLock()
		s := sm.dataTrackSubscriptions["track"]
		sm.lock.RUnlock()
		require.Eventually(t, func() bool {
			return !s.needsSubscribe()
		}, subSettleTimeout, subCheckInterval, "should be subscribed")

		// the client subscribes again while the removal runs,
		// then the down track closes, as DataTrack.RemoveSubscriber does
		ddt := s.getDataDownTrack().(*typesfakes.FakeDataDownTrack)
		resolver.dataTrack.RemoveSubscriberCalls(func(livekit.ParticipantID) {
			sm.SubscribeToDataTrack("track")
			ddt.OnCloseArgsForCall(0)()
		})
		sm.UnsubscribeFromDataTrack("track")

		require.Eventually(t, func() bool {
			return resolver.dataTrack.AddSubscriberCallCount() == 2 && s.getDataDownTrack() != nil
		}, subSettleTimeout, subCheckInterval, "should be subscribed again")
		sm.lock.RLock()
		require.Same(t, s, sm.dataTrackSubscriptions["track"])
		sm.lock.RUnlock()
	})

	t.Run("unsubscribe before data track resolves", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		// no track available, subscribe attempts fail with ErrTrackNotFound
		resolver := newTestDataTrackResolver(true, false, "pub", "pubID")
		sm.params.DataTrackResolver = resolver.Resolve

		sm.SubscribeToDataTrack("track")
		sm.lock.RLock()
		s := sm.dataTrackSubscriptions["track"]
		sm.lock.RUnlock()
		require.NotNil(t, s)

		// let the worker attempt (and fail) the subscribe
		require.Eventually(t, func() bool {
			return s.getNumAttempts() > 0
		}, subSettleTimeout, subCheckInterval, "no subscribe attempt was made")

		sm.UnsubscribeFromDataTrack("track")
		require.Eventually(t, func() bool {
			sm.lock.RLock()
			_, ok := sm.dataTrackSubscriptions["track"]
			sm.lock.RUnlock()
			return !ok
		}, subSettleTimeout, subCheckInterval, "data track subscription was not cleaned up")
	})
}

func TestSubscribePacerTakesVideoTokensOnly(t *testing.T) {
	for _, tc := range []struct {
		name        string
		kind        livekit.TrackType
		isSync      bool
		setupFails  bool
		tokenIsUsed bool
	}{
		{"video", livekit.TrackType_VIDEO, false, false, true},
		{"audio", livekit.TrackType_AUDIO, false, false, false},
		{"video one-shot", livekit.TrackType_VIDEO, true, false, false},
		// a failed setup spends its token, like a real pacer that gates before setup
		{"video setup fails", livekit.TrackType_VIDEO, false, true, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			// a rate this low holds one token and takes 1000 s to refill it, so after one
			// take the next subscribe waits however slow the test runs
			pacer := NewSubscribePacer([]float64{0.001})
			sm := newTestSubscriptionManager()
			defer sm.Close(false)
			sm.params.SubscribePacer = pacer
			attempted := atomic.Bool{}
			resolver := newTestResolver(true, true, "pub", "pubID")
			sm.params.TrackResolver = func(sub types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult {
				res := resolver.Resolve(sub, trackID)
				mt := res.Track.(*typesfakes.FakeMediaTrack)
				mt.KindReturns(tc.kind)
				if tc.setupFails {
					mt.AddSubscriberCalls(func(types.LocalParticipant) (types.SubscribedTrack, error) {
						attempted.Store(true)
						return nil, ErrNoReceiver
					})
				}
				return res
			}
			sm.params.OnTrackSubscribed = func(subTrack types.SubscribedTrack) {
				attempted.Store(true)
			}

			sm.SubscribeToTrack("track", tc.isSync)
			require.Eventually(t, attempted.Load, subSettleTimeout, subCheckInterval, "subscribe was not attempted")

			require.Equal(t, tc.tokenIsUsed, pacer.Delays(time.Now())["0.001"] > 0)
		})
	}
}

// A canSubscribe revoke that lands while a subscribe is in flight cannot see the subscription
// yet. The participant here allows the subscribe at its start and is revoked by the time the
// subscription is recorded, so the subscribe must undo itself.
func TestSubscribeRevokedInFlight(t *testing.T) {
	revokeAfterFirstCheck := func(sm *SubscriptionManager) {
		p := sm.params.Participant.(*typesfakes.FakeLocalParticipant)
		p.CanSubscribeReturnsOnCall(0, true)
		p.CanSubscribeReturns(false)
	}

	t.Run("media", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		revokeAfterFirstCheck(sm)

		resolver := newTestResolver(true, true, "pub", "pubID")
		var lock sync.Mutex
		var tracks []*typesfakes.FakeMediaTrack
		sm.params.TrackResolver = func(sub types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult {
			res := resolver.Resolve(sub, trackID)
			lock.Lock()
			tracks = append(tracks, res.Track.(*typesfakes.FakeMediaTrack))
			lock.Unlock()
			return res
		}
		subscribed := atomic.Bool{}
		sm.params.OnTrackSubscribed = func(subTrack types.SubscribedTrack) {
			subscribed.Store(true)
		}

		sm.SubscribeToTrack("track", false)
		require.Eventually(t, func() bool {
			lock.Lock()
			defer lock.Unlock()
			return len(tracks) != 0 && tracks[0].RemoveSubscriberCallCount() == 1
		}, subSettleTimeout, subCheckInterval, "the subscriber should be removed")

		lock.Lock()
		subID, isExpectedToResume := tracks[0].RemoveSubscriberArgsForCall(0)
		lock.Unlock()
		require.Equal(t, sm.params.Participant.ID(), subID)
		require.False(t, isExpectedToResume)
		require.Empty(t, sm.GetSubscribedParticipants())
		require.False(t, subscribed.Load())
	})

	t.Run("data", func(t *testing.T) {
		sm := newTestSubscriptionManager()
		defer sm.Close(false)
		revokeAfterFirstCheck(sm)

		resolver := newTestDataTrackResolver(true, true, "pub", "pubID")
		sm.params.DataTrackResolver = resolver.Resolve

		sm.SubscribeToDataTrack("track")
		require.Eventually(t, func() bool {
			return resolver.dataTrack.RemoveSubscriberCallCount() == 1
		}, subSettleTimeout, subCheckInterval, "the subscriber should be removed")

		require.Equal(t, sm.params.Participant.ID(), resolver.dataTrack.RemoveSubscriberArgsForCall(0))
		require.Empty(t, sm.GetSubscribedParticipants())
	})
}

type testSubscriptionParams struct {
	SubscriptionLimitAudio int32
	SubscriptionLimitVideo int32
}

func newTestSubscriptionManager() *SubscriptionManager {
	return newTestSubscriptionManagerWithParams(testSubscriptionParams{})
}

func newTestSubscriptionManagerWithParams(params testSubscriptionParams) *SubscriptionManager {
	p := &typesfakes.FakeLocalParticipant{}
	p.CanSubscribeReturns(true)
	p.IDReturns("subID")
	p.IdentityReturns("sub")
	p.KindReturns(livekit.ParticipantInfo_STANDARD)

	tl := &typesfakes.FakeParticipantTelemetryListener{}
	p.GetTelemetryListenerReturns(tl)

	return NewSubscriptionManager(SubscriptionManagerParams{
		Participant:         p,
		Logger:              logger.GetLogger(),
		OnTrackSubscribed:   func(subTrack types.SubscribedTrack) {},
		OnTrackUnsubscribed: func(subTrack types.SubscribedTrack) {},
		OnSubscriptionError: func(trackID livekit.TrackID, fatal bool, err error) {},
		TrackResolver: func(sub types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult {
			return types.MediaResolverResult{}
		},
		SubscriptionLimitAudio: params.SubscriptionLimitAudio,
		SubscriptionLimitVideo: params.SubscriptionLimitVideo,
	})
}

type testResolver struct {
	lock          sync.Mutex
	hasPermission bool
	hasTrack      bool
	pubIdentity   livekit.ParticipantIdentity
	pubID         livekit.ParticipantID

	paused bool
}

func newTestResolver(hasPermission bool, hasTrack bool, pubIdentity livekit.ParticipantIdentity, pubID livekit.ParticipantID) *testResolver {
	return &testResolver{
		hasPermission: hasPermission,
		hasTrack:      hasTrack,
		pubIdentity:   pubIdentity,
		pubID:         pubID,
	}
}

func (t *testResolver) SetPause(paused bool) {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.paused = paused
}

func (t *testResolver) Resolve(_subscriber types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult {
	t.lock.Lock()
	defer t.lock.Unlock()
	res := types.MediaResolverResult{
		TrackChangedNotifier: utils.NewChangeNotifier(),
		TrackRemovedNotifier: utils.NewChangeNotifier(),
		HasPermission:        t.hasPermission,
		PublisherID:          t.pubID,
		PublisherIdentity:    t.pubIdentity,
	}
	if t.hasTrack && !t.paused {
		mt := &typesfakes.FakeMediaTrack{}
		st := &typesfakes.FakeSubscribedTrack{}
		st.IDReturns(trackID)
		st.PublisherIDReturns(t.pubID)
		st.PublisherIdentityReturns(t.pubIdentity)
		mt.AddSubscriberCalls(func(sub types.LocalParticipant) (types.SubscribedTrack, error) {
			st.SubscriberReturns(sub)
			return st, nil
		})
		st.MediaTrackReturns(mt)
		res.Track = mt
	}
	return res
}

type testDataTrackResolver struct {
	lock          sync.Mutex
	hasPermission bool
	hasTrack      bool
	pubIdentity   livekit.ParticipantIdentity
	pubID         livekit.ParticipantID

	dataTrack *typesfakes.FakeDataTrack
}

func newTestDataTrackResolver(hasPermission bool, hasTrack bool, pubIdentity livekit.ParticipantIdentity, pubID livekit.ParticipantID) *testDataTrackResolver {
	r := &testDataTrackResolver{
		hasPermission: hasPermission,
		hasTrack:      hasTrack,
		pubIdentity:   pubIdentity,
		pubID:         pubID,
		dataTrack:     &typesfakes.FakeDataTrack{},
	}
	r.dataTrack.PublisherIDReturns(pubID)
	r.dataTrack.PublisherIdentityReturns(pubIdentity)
	r.dataTrack.AddSubscriberCalls(func(sub types.LocalParticipant) (types.DataDownTrack, error) {
		ddt := &typesfakes.FakeDataDownTrack{}
		ddt.PublishDataTrackReturns(r.dataTrack)
		return ddt, nil
	})
	return r
}

func (t *testDataTrackResolver) SetHasPermission(hasPermission bool) {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.hasPermission = hasPermission
}

func (t *testDataTrackResolver) Resolve(_subscriber types.LocalParticipant, trackID livekit.TrackID) types.DataResolverResult {
	t.lock.Lock()
	defer t.lock.Unlock()
	res := types.DataResolverResult{
		TrackChangedNotifier: utils.NewChangeNotifier(),
		TrackRemovedNotifier: utils.NewChangeNotifier(),
		HasPermission:        t.hasPermission,
		PublisherID:          t.pubID,
		PublisherIdentity:    t.pubIdentity,
	}
	if t.hasTrack {
		t.dataTrack.IDReturns(trackID)
		res.DataTrack = t.dataTrack
	}
	return res
}

func setTestSubscribedTrackBound(t *testing.T, st types.SubscribedTrack) {
	fst, ok := st.(*typesfakes.FakeSubscribedTrack)
	require.True(t, ok)

	for i := 0; i < fst.AddOnBindCallCount(); i++ {
		fst.AddOnBindArgsForCall(i)(nil)
	}
}

func setTestSubscribedTrackClosed(t *testing.T, st types.SubscribedTrack, isExpectedToResume bool) {
	fst, ok := st.(*typesfakes.FakeSubscribedTrack)
	require.True(t, ok)

	fst.OnCloseArgsForCall(0)(isExpectedToResume)
}

func setTestDataDownTrackClosed(t *testing.T, ddt types.DataDownTrack) {
	fddt, ok := ddt.(*typesfakes.FakeDataDownTrack)
	require.True(t, ok)

	fddt.OnCloseArgsForCall(0)()
}