livekit/pkg/rtc/subscribepacer.go

// Copyright 2026 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 (
	"slices"
	"strconv"
	"sync"
	"time"

	"github.com/livekit/protocol/logger"
)

// SubscribePacer measures, in shadow, how long video subscribes on this node would wait
// if they were paced at each of a set of rates. It never delays a subscribe.
//
// Each rate is a token bucket that refills at that many subscribes per second and holds
// one second of tokens, so a burst of up to one second's worth goes at once and the rest
// waits in line.
type SubscribePacer struct {
	lock    sync.Mutex
	buckets []subscribePaceBucket
}

type subscribePaceBucket struct {
	rateLabel string
	interval  time.Duration // time to refill one token
	burst     time.Duration // how far ahead of now tokens may be taken without waiting
	nextAt    time.Time     // when the next token becomes free, if every earlier subscribe had waited for its token
}

// NewSubscribePacer returns nil when no rates are given, which turns the shadow off.
func NewSubscribePacer(rates []float64) *SubscribePacer {
	rates = slices.Clone(rates)
	slices.Sort(rates)

	var buckets []subscribePaceBucket
	var labels []string
	for i, rate := range rates {
		if rate <= 0 || (i > 0 && rate == rates[i-1]) {
			continue
		}

		interval := time.Duration(float64(time.Second) / rate)
		label := strconv.FormatFloat(rate, 'f', -1, 64)
		buckets = append(buckets, subscribePaceBucket{
			rateLabel: label,
			interval:  interval,
			burst:     max(time.Second-interval, 0),
		})
		labels = append(labels, label)
	}
	if len(buckets) == 0 {
		return nil
	}

	logger.Infow("subscribe pacing shadow enabled", "rates", labels)
	return &SubscribePacer{buckets: buckets}
}

// Delays takes a token from every bucket for a subscribe that starts at now and returns,
// keyed by rate, how long that subscribe would have waited for it.
func (p *SubscribePacer) Delays(now time.Time) map[string]time.Duration {
	if p == nil {
		return nil
	}

	p.lock.Lock()
	defer p.lock.Unlock()

	delays := make(map[string]time.Duration, len(p.buckets))
	for i := range p.buckets {
		b := &p.buckets[i]
		takeAt := b.nextAt
		if takeAt.Before(now) {
			takeAt = now
		}
		delays[b.rateLabel] = max(takeAt.Sub(now)-b.burst, 0)
		b.nextAt = takeAt.Add(b.interval)
	}
	return delays
}