Add control of playout delay (#1838)

* Add control of playout delay

Add config to enable playout delay. The delay will be limited by
[min,max] in the config option and calculated by upstream & downstream
RTT.

* check protocol version to enable playout delay

* Move config to room, limit playout-delay update interval, solve comments

* Remove adaptive playout-delay

* Remove unused config

config-sample.yaml

@@ -152,6 +152,11 @@ keys:
 #   enable_remote_unmute: true
 #   # limit size of room and participant's metadata, 0 for no limit
 #   max_metadata_size: 0
+#   # control playout delay in ms of video track (and associated audio track) 
+#   playout_delay:
+#     enabled: true
+#     min: 100
+#     max: 300
 
 # Webhooks
 # when configured, LiveKit notifies your URL handler with room events

pkg/config/config.go

@@ -197,6 +197,12 @@ type StreamTrackersConfig struct {
 	Screenshare StreamTrackerConfig `yaml:"screenshare,omitempty"`
 }
 
+type PlayoutDelayConfig struct {
+	Enabled bool `yaml:"enabled,omitempty"`
+	Min     int  `yaml:"min,omitempty"`
+	Max     int  `yaml:"max,omitempty"`
+}
+
 type VideoConfig struct {
 	DynacastPauseDelay time.Duration        `yaml:"dynacast_pause_delay,omitempty"`
 	StreamTracker      StreamTrackersConfig `yaml:"stream_tracker,omitempty"`
@@ -204,12 +210,13 @@ type VideoConfig struct {
 
 type RoomConfig struct {
 	// enable rooms to be automatically created
-	AutoCreate         bool        `yaml:"auto_create,omitempty"`
-	EnabledCodecs      []CodecSpec `yaml:"enabled_codecs,omitempty"`
-	MaxParticipants    uint32      `yaml:"max_participants,omitempty"`
-	EmptyTimeout       uint32      `yaml:"empty_timeout,omitempty"`
-	EnableRemoteUnmute bool        `yaml:"enable_remote_unmute,omitempty"`
-	MaxMetadataSize    uint32      `yaml:"max_metadata_size,omitempty"`
+	AutoCreate         bool               `yaml:"auto_create,omitempty"`
+	EnabledCodecs      []CodecSpec        `yaml:"enabled_codecs,omitempty"`
+	MaxParticipants    uint32             `yaml:"max_participants,omitempty"`
+	EmptyTimeout       uint32             `yaml:"empty_timeout,omitempty"`
+	EnableRemoteUnmute bool               `yaml:"enable_remote_unmute,omitempty"`
+	MaxMetadataSize    uint32             `yaml:"max_metadata_size,omitempty"`
+	PlayoutDelay       PlayoutDelayConfig `yaml:"playout_delay,omitempty"`
 }
 
 type CodecSpec struct {

pkg/rtc/config.go

@@ -86,7 +86,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) {
 				sdp.SDESRTPStreamIDURI,
 				sdp.TransportCCURI,
 				frameMarking,
-				dd.ExtensionUrl,
+				dd.ExtensionURI,
 			},
 		},
 		RTCPFeedback: RTCPFeedbackConfig{
@@ -106,7 +106,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) {
 	subscriberConfig := DirectionConfig{
 		StrictACKs: conf.RTC.StrictACKs,
 		RTPHeaderExtension: RTPHeaderExtensionConfig{
-			Video: []string{dd.ExtensionUrl},
+			Video: []string{dd.ExtensionURI},
 		},
 		RTCPFeedback: RTCPFeedbackConfig{
 			Video: []webrtc.RTCPFeedback{

pkg/rtc/mediatrackreceiver.go

@@ -226,7 +226,7 @@ func (t *MediaTrackReceiver) SetPotentialCodecs(codecs []webrtc.RTPCodecParamete
 	// that is munged in svc codec.
 	headersWithoutDD := make([]webrtc.RTPHeaderExtensionParameter, 0, len(headers))
 	for _, h := range headers {
-		if h.URI != dependencydescriptor.ExtensionUrl {
+		if h.URI != dependencydescriptor.ExtensionURI {
 			headersWithoutDD = append(headersWithoutDD, h)
 		}
 	}
@@ -401,6 +401,13 @@ func (t *MediaTrackReceiver) Source() livekit.TrackSource {
 	return t.trackInfo.Source
 }
 
+func (t *MediaTrackReceiver) Stream() string {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+
+	return t.trackInfo.Stream
+}
+
 func (t *MediaTrackReceiver) PublisherID() livekit.ParticipantID {
 	return t.params.ParticipantID
 }

pkg/rtc/mediatracksubscriptions.go

@@ -112,20 +112,29 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *
 	for _, c := range codecs {
 		c.RTCPFeedback = rtcpFeedback
 	}
+
+	streamID := wr.StreamID()
+	if sub.SupportSyncStreamID() && t.params.MediaTrack.Stream() != "" {
+		streamID = PackSyncStreamID(t.params.MediaTrack.PublisherID(), t.params.MediaTrack.Stream())
+	}
+
 	var trailer []byte
 	if t.params.MediaTrack.IsEncrypted() {
 		trailer = sub.GetTrailer()
 	}
-	downTrack, err := sfu.NewDownTrack(
-		codecs,
-		wr,
-		sub.GetBufferFactory(),
-		subscriberID,
-		t.params.ReceiverConfig.PacketBufferSize,
-		sub.GetPacer(),
-		trailer,
-		LoggerWithTrack(sub.GetLogger(), trackID, t.params.IsRelayed),
-	)
+
+	downTrack, err := sfu.NewDownTrack(sfu.DowntrackParams{
+		Codecs:            codecs,
+		Receiver:          wr,
+		BufferFactory:     sub.GetBufferFactory(),
+		SubID:             subscriberID,
+		StreamID:          streamID,
+		MaxTrack:          t.params.ReceiverConfig.PacketBufferSize,
+		PlayoutDelayLimit: sub.GetPlayoutDelayConfig(),
+		Pacer:             sub.GetPacer(),
+		Trailer:           trailer,
+		Logger:            LoggerWithTrack(sub.GetLogger(), trackID, t.params.IsRelayed),
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -200,6 +209,8 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *
 		reusingTransceiver.Store(true)
 		rtpSender := existingTransceiver.Sender()
 		if rtpSender != nil {
+			// replaced track will bind immediately without negotiation, SetTransceiver first before bind
+			downTrack.SetTransceiver(existingTransceiver)
 			err := rtpSender.ReplaceTrack(downTrack)
 			if err == nil {
 				sender = rtpSender
@@ -260,9 +271,6 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *
 	subTrack.SetNeedsNegotiation(!replacedTrack)
 	subTrack.SetRTPSender(sender)
 
-	sendParameters := sender.GetParameters()
-	downTrack.SetRTPHeaderExtensions(sendParameters.HeaderExtensions)
-
 	downTrack.SetTransceiver(transceiver)
 
 	downTrack.OnCloseHandler(func(willBeResumed bool) {

pkg/rtc/participant.go

@@ -108,6 +108,7 @@ type ParticipantParams struct {
 	SubscriberAllowPause         bool
 	SubscriptionLimitAudio       int32
 	SubscriptionLimitVideo       int32
+	PlayoutDelay                 *livekit.PlayoutDelay
 }
 
 type ParticipantImpl struct {
@@ -1104,6 +1105,7 @@ func (p *ParticipantImpl) setupTransportManager() error {
 		TCPFallbackRTTThreshold:  p.params.TCPFallbackRTTThreshold,
 		AllowUDPUnstableFallback: p.params.AllowUDPUnstableFallback,
 		TURNSEnabled:             p.params.TURNSEnabled,
+		AllowPlayoutDelay:        p.params.PlayoutDelay.GetEnabled() && p.SupportSyncStreamID(),
 		Logger:                   p.params.Logger,
 	})
 	if err != nil {
@@ -1584,6 +1586,10 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l
 		DisableRed: req.DisableRed,
 		Stereo:     req.Stereo,
 		Encryption: req.Encryption,
+		Stream:     req.Stream,
+	}
+	if ti.Stream == "" {
+		ti.Stream = StreamFromTrackSource(ti.Source)
 	}
 	p.setStableTrackID(req.Cid, ti)
 	for _, codec := range req.SimulcastCodecs {
@@ -2214,6 +2220,14 @@ func (p *ParticipantImpl) UpdateMediaLoss(nodeID livekit.NodeID, trackID livekit
 	return nil
 }
 
+func (p *ParticipantImpl) GetPlayoutDelayConfig() *livekit.PlayoutDelay {
+	return p.params.PlayoutDelay
+}
+
+func (p *ParticipantImpl) SupportSyncStreamID() bool {
+	return p.ProtocolVersion().SupportSyncStreamID() && !p.params.ClientInfo.isFirefox()
+}
+
 func codecsFromMediaDescription(m *sdp.MediaDescription) (out []sdp.Codec, err error) {
 	s := &sdp.SessionDescription{
 		MediaDescriptions: []*sdp.MediaDescription{m},

pkg/rtc/participant_sdp.go

@@ -155,7 +155,7 @@ func (p *ParticipantImpl) setCodecPreferencesVideoForPublisher(offer webrtc.Sess
 		// remove dd extension if av1/vp9 not preferred
 		if !strings.Contains(strings.ToLower(mime), "av1") && !strings.Contains(strings.ToLower(mime), "vp9") {
 			for i, attr := range unmatchVideo.Attributes {
-				if strings.Contains(attr.Value, dd.ExtensionUrl) {
+				if strings.Contains(attr.Value, dd.ExtensionURI) {
 					unmatchVideo.Attributes[i] = unmatchVideo.Attributes[len(unmatchVideo.Attributes)-1]
 					unmatchVideo.Attributes = unmatchVideo.Attributes[:len(unmatchVideo.Attributes)-1]
 					break

pkg/rtc/transport.go

@@ -43,6 +43,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu/pacer"
+	"github.com/livekit/livekit-server/pkg/sfu/rtpextension"
 	"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
@@ -251,10 +252,17 @@ type TransportParams struct {
 	ClientInfo              ClientInfo
 	IsOfferer               bool
 	IsSendSide              bool
+	AllowPlayoutDelay       bool
 }
 
 func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimator cc.BandwidthEstimator)) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) {
-	me, err := createMediaEngine(params.EnabledCodecs, params.DirectionConfig)
+	directionConfig := params.DirectionConfig
+
+	if params.AllowPlayoutDelay {
+		directionConfig.RTPHeaderExtension.Video = append(directionConfig.RTPHeaderExtension.Video, rtpextension.PlayoutDelayURI)
+	}
+
+	me, err := createMediaEngine(params.EnabledCodecs, directionConfig)
 	if err != nil {
 		return nil, nil, err
 	}

pkg/rtc/transportmanager.go

@@ -65,6 +65,7 @@ type TransportManagerParams struct {
 	TCPFallbackRTTThreshold  int
 	AllowUDPUnstableFallback bool
 	TURNSEnabled             bool
+	AllowPlayoutDelay        bool
 	Logger                   logger.Logger
 }
 
@@ -184,6 +185,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
 		ClientInfo:              params.ClientInfo,
 		IsOfferer:               true,
 		IsSendSide:              true,
+		AllowPlayoutDelay:       params.AllowPlayoutDelay,
 	})
 	if err != nil {
 		return nil, err

pkg/rtc/types/interfaces.go

@@ -295,6 +295,7 @@ type LocalParticipant interface {
 	GetLogger() logger.Logger
 	GetAdaptiveStream() bool
 	ProtocolVersion() ProtocolVersion
+	SupportSyncStreamID() bool
 	ConnectedAt() time.Time
 	IsClosed() bool
 	IsReady() bool
@@ -305,6 +306,7 @@ type LocalParticipant interface {
 	GetClientConfiguration() *livekit.ClientConfiguration
 	GetICEConnectionType() ICEConnectionType
 	GetBufferFactory() *buffer.Factory
+	GetPlayoutDelayConfig() *livekit.PlayoutDelay
 
 	SetResponseSink(sink routing.MessageSink)
 	CloseSignalConnection(reason SignallingCloseReason)
@@ -429,6 +431,7 @@ type MediaTrack interface {
 	Kind() livekit.TrackType
 	Name() string
 	Source() livekit.TrackSource
+	Stream() string
 
 	ToProto() *livekit.TrackInfo
 

pkg/rtc/types/protocol_version.go

@@ -16,7 +16,7 @@ package types
 
 type ProtocolVersion int
 
-const CurrentProtocol = 9
+const CurrentProtocol = 10
 
 func (v ProtocolVersion) SupportsPackedStreamId() bool {
 	return v > 0
@@ -71,3 +71,7 @@ func (v ProtocolVersion) SupportFastStart() bool {
 func (v ProtocolVersion) SupportHandlesDisconnectedUpdate() bool {
 	return v > 8
 }
+
+func (v ProtocolVersion) SupportSyncStreamID() bool {
+	return v > 9
+}

pkg/rtc/types/typesfakes/fake_local_media_track.go

@@ -302,6 +302,16 @@ type FakeLocalMediaTrack struct {
 	sourceReturnsOnCall map[int]struct {
 		result1 livekit.TrackSource
 	}
+	StreamStub        func() string
+	streamMutex       sync.RWMutex
+	streamArgsForCall []struct {
+	}
+	streamReturns struct {
+		result1 string
+	}
+	streamReturnsOnCall map[int]struct {
+		result1 string
+	}
 	ToProtoStub        func() *livekit.TrackInfo
 	toProtoMutex       sync.RWMutex
 	toProtoArgsForCall []struct {
@@ -1893,6 +1903,59 @@ func (fake *FakeLocalMediaTrack) SourceReturnsOnCall(i int, result1 livekit.Trac
 	}{result1}
 }
 
+func (fake *FakeLocalMediaTrack) Stream() string {
+	fake.streamMutex.Lock()
+	ret, specificReturn := fake.streamReturnsOnCall[len(fake.streamArgsForCall)]
+	fake.streamArgsForCall = append(fake.streamArgsForCall, struct {
+	}{})
+	stub := fake.StreamStub
+	fakeReturns := fake.streamReturns
+	fake.recordInvocation("Stream", []interface{}{})
+	fake.streamMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeLocalMediaTrack) StreamCallCount() int {
+	fake.streamMutex.RLock()
+	defer fake.streamMutex.RUnlock()
+	return len(fake.streamArgsForCall)
+}
+
+func (fake *FakeLocalMediaTrack) StreamCalls(stub func() string) {
+	fake.streamMutex.Lock()
+	defer fake.streamMutex.Unlock()
+	fake.StreamStub = stub
+}
+
+func (fake *FakeLocalMediaTrack) StreamReturns(result1 string) {
+	fake.streamMutex.Lock()
+	defer fake.streamMutex.Unlock()
+	fake.StreamStub = nil
+	fake.streamReturns = struct {
+		result1 string
+	}{result1}
+}
+
+func (fake *FakeLocalMediaTrack) StreamReturnsOnCall(i int, result1 string) {
+	fake.streamMutex.Lock()
+	defer fake.streamMutex.Unlock()
+	fake.StreamStub = nil
+	if fake.streamReturnsOnCall == nil {
+		fake.streamReturnsOnCall = make(map[int]struct {
+			result1 string
+		})
+	}
+	fake.streamReturnsOnCall[i] = struct {
+		result1 string
+	}{result1}
+}
+
 func (fake *FakeLocalMediaTrack) ToProto() *livekit.TrackInfo {
 	fake.toProtoMutex.Lock()
 	ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)]
@@ -2050,6 +2113,8 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} {
 	defer fake.signalCidMutex.RUnlock()
 	fake.sourceMutex.RLock()
 	defer fake.sourceMutex.RUnlock()
+	fake.streamMutex.RLock()
+	defer fake.streamMutex.RUnlock()
 	fake.toProtoMutex.RLock()
 	defer fake.toProtoMutex.RUnlock()
 	fake.updateVideoLayersMutex.RLock()

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -263,6 +263,16 @@ type FakeLocalParticipant struct {
 	getPacerReturnsOnCall map[int]struct {
 		result1 pacer.Pacer
 	}
+	GetPlayoutDelayConfigStub        func() *livekit.PlayoutDelay
+	getPlayoutDelayConfigMutex       sync.RWMutex
+	getPlayoutDelayConfigArgsForCall []struct {
+	}
+	getPlayoutDelayConfigReturns struct {
+		result1 *livekit.PlayoutDelay
+	}
+	getPlayoutDelayConfigReturnsOnCall map[int]struct {
+		result1 *livekit.PlayoutDelay
+	}
 	GetPublishedTrackStub        func(livekit.TrackID) types.MediaTrack
 	getPublishedTrackMutex       sync.RWMutex
 	getPublishedTrackArgsForCall []struct {
@@ -772,6 +782,16 @@ type FakeLocalParticipant struct {
 		arg2 livekit.TrackID
 		arg3 bool
 	}
+	SupportSyncStreamIDStub        func() bool
+	supportSyncStreamIDMutex       sync.RWMutex
+	supportSyncStreamIDArgsForCall []struct {
+	}
+	supportSyncStreamIDReturns struct {
+		result1 bool
+	}
+	supportSyncStreamIDReturnsOnCall map[int]struct {
+		result1 bool
+	}
 	ToProtoStub        func() *livekit.ParticipantInfo
 	toProtoMutex       sync.RWMutex
 	toProtoArgsForCall []struct {
@@ -2149,6 +2169,59 @@ func (fake *FakeLocalParticipant) GetPacerReturnsOnCall(i int, result1 pacer.Pac
 	}{result1}
 }
 
+func (fake *FakeLocalParticipant) GetPlayoutDelayConfig() *livekit.PlayoutDelay {
+	fake.getPlayoutDelayConfigMutex.Lock()
+	ret, specificReturn := fake.getPlayoutDelayConfigReturnsOnCall[len(fake.getPlayoutDelayConfigArgsForCall)]
+	fake.getPlayoutDelayConfigArgsForCall = append(fake.getPlayoutDelayConfigArgsForCall, struct {
+	}{})
+	stub := fake.GetPlayoutDelayConfigStub
+	fakeReturns := fake.getPlayoutDelayConfigReturns
+	fake.recordInvocation("GetPlayoutDelayConfig", []interface{}{})
+	fake.getPlayoutDelayConfigMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeLocalParticipant) GetPlayoutDelayConfigCallCount() int {
+	fake.getPlayoutDelayConfigMutex.RLock()
+	defer fake.getPlayoutDelayConfigMutex.RUnlock()
+	return len(fake.getPlayoutDelayConfigArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) GetPlayoutDelayConfigCalls(stub func() *livekit.PlayoutDelay) {
+	fake.getPlayoutDelayConfigMutex.Lock()
+	defer fake.getPlayoutDelayConfigMutex.Unlock()
+	fake.GetPlayoutDelayConfigStub = stub
+}
+
+func (fake *FakeLocalParticipant) GetPlayoutDelayConfigReturns(result1 *livekit.PlayoutDelay) {
+	fake.getPlayoutDelayConfigMutex.Lock()
+	defer fake.getPlayoutDelayConfigMutex.Unlock()
+	fake.GetPlayoutDelayConfigStub = nil
+	fake.getPlayoutDelayConfigReturns = struct {
+		result1 *livekit.PlayoutDelay
+	}{result1}
+}
+
+func (fake *FakeLocalParticipant) GetPlayoutDelayConfigReturnsOnCall(i int, result1 *livekit.PlayoutDelay) {
+	fake.getPlayoutDelayConfigMutex.Lock()
+	defer fake.getPlayoutDelayConfigMutex.Unlock()
+	fake.GetPlayoutDelayConfigStub = nil
+	if fake.getPlayoutDelayConfigReturnsOnCall == nil {
+		fake.getPlayoutDelayConfigReturnsOnCall = make(map[int]struct {
+			result1 *livekit.PlayoutDelay
+		})
+	}
+	fake.getPlayoutDelayConfigReturnsOnCall[i] = struct {
+		result1 *livekit.PlayoutDelay
+	}{result1}
+}
+
 func (fake *FakeLocalParticipant) GetPublishedTrack(arg1 livekit.TrackID) types.MediaTrack {
 	fake.getPublishedTrackMutex.Lock()
 	ret, specificReturn := fake.getPublishedTrackReturnsOnCall[len(fake.getPublishedTrackArgsForCall)]
@@ -5008,6 +5081,59 @@ func (fake *FakeLocalParticipant) SubscriptionPermissionUpdateArgsForCall(i int)
 	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
 }
 
+func (fake *FakeLocalParticipant) SupportSyncStreamID() bool {
+	fake.supportSyncStreamIDMutex.Lock()
+	ret, specificReturn := fake.supportSyncStreamIDReturnsOnCall[len(fake.supportSyncStreamIDArgsForCall)]
+	fake.supportSyncStreamIDArgsForCall = append(fake.supportSyncStreamIDArgsForCall, struct {
+	}{})
+	stub := fake.SupportSyncStreamIDStub
+	fakeReturns := fake.supportSyncStreamIDReturns
+	fake.recordInvocation("SupportSyncStreamID", []interface{}{})
+	fake.supportSyncStreamIDMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeLocalParticipant) SupportSyncStreamIDCallCount() int {
+	fake.supportSyncStreamIDMutex.RLock()
+	defer fake.supportSyncStreamIDMutex.RUnlock()
+	return len(fake.supportSyncStreamIDArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) SupportSyncStreamIDCalls(stub func() bool) {
+	fake.supportSyncStreamIDMutex.Lock()
+	defer fake.supportSyncStreamIDMutex.Unlock()
+	fake.SupportSyncStreamIDStub = stub
+}
+
+func (fake *FakeLocalParticipant) SupportSyncStreamIDReturns(result1 bool) {
+	fake.supportSyncStreamIDMutex.Lock()
+	defer fake.supportSyncStreamIDMutex.Unlock()
+	fake.SupportSyncStreamIDStub = nil
+	fake.supportSyncStreamIDReturns = struct {
+		result1 bool
+	}{result1}
+}
+
+func (fake *FakeLocalParticipant) SupportSyncStreamIDReturnsOnCall(i int, result1 bool) {
+	fake.supportSyncStreamIDMutex.Lock()
+	defer fake.supportSyncStreamIDMutex.Unlock()
+	fake.SupportSyncStreamIDStub = nil
+	if fake.supportSyncStreamIDReturnsOnCall == nil {
+		fake.supportSyncStreamIDReturnsOnCall = make(map[int]struct {
+			result1 bool
+		})
+	}
+	fake.supportSyncStreamIDReturnsOnCall[i] = struct {
+		result1 bool
+	}{result1}
+}
+
 func (fake *FakeLocalParticipant) ToProto() *livekit.ParticipantInfo {
 	fake.toProtoMutex.Lock()
 	ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)]
@@ -5703,6 +5829,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.getLoggerMutex.RUnlock()
 	fake.getPacerMutex.RLock()
 	defer fake.getPacerMutex.RUnlock()
+	fake.getPlayoutDelayConfigMutex.RLock()
+	defer fake.getPlayoutDelayConfigMutex.RUnlock()
 	fake.getPublishedTrackMutex.RLock()
 	defer fake.getPublishedTrackMutex.RUnlock()
 	fake.getPublishedTracksMutex.RLock()
@@ -5831,6 +5959,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.subscriptionPermissionMutex.RUnlock()
 	fake.subscriptionPermissionUpdateMutex.RLock()
 	defer fake.subscriptionPermissionUpdateMutex.RUnlock()
+	fake.supportSyncStreamIDMutex.RLock()
+	defer fake.supportSyncStreamIDMutex.RUnlock()
 	fake.toProtoMutex.RLock()
 	defer fake.toProtoMutex.RUnlock()
 	fake.toProtoWithVersionMutex.RLock()

pkg/rtc/types/typesfakes/fake_media_track.go

@@ -236,6 +236,16 @@ type FakeMediaTrack struct {
 	sourceReturnsOnCall map[int]struct {
 		result1 livekit.TrackSource
 	}
+	StreamStub        func() string
+	streamMutex       sync.RWMutex
+	streamArgsForCall []struct {
+	}
+	streamReturns struct {
+		result1 string
+	}
+	streamReturnsOnCall map[int]struct {
+		result1 string
+	}
 	ToProtoStub        func() *livekit.TrackInfo
 	toProtoMutex       sync.RWMutex
 	toProtoArgsForCall []struct {
@@ -1474,6 +1484,59 @@ func (fake *FakeMediaTrack) SourceReturnsOnCall(i int, result1 livekit.TrackSour
 	}{result1}
 }
 
+func (fake *FakeMediaTrack) Stream() string {
+	fake.streamMutex.Lock()
+	ret, specificReturn := fake.streamReturnsOnCall[len(fake.streamArgsForCall)]
+	fake.streamArgsForCall = append(fake.streamArgsForCall, struct {
+	}{})
+	stub := fake.StreamStub
+	fakeReturns := fake.streamReturns
+	fake.recordInvocation("Stream", []interface{}{})
+	fake.streamMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeMediaTrack) StreamCallCount() int {
+	fake.streamMutex.RLock()
+	defer fake.streamMutex.RUnlock()
+	return len(fake.streamArgsForCall)
+}
+
+func (fake *FakeMediaTrack) StreamCalls(stub func() string) {
+	fake.streamMutex.Lock()
+	defer fake.streamMutex.Unlock()
+	fake.StreamStub = stub
+}
+
+func (fake *FakeMediaTrack) StreamReturns(result1 string) {
+	fake.streamMutex.Lock()
+	defer fake.streamMutex.Unlock()
+	fake.StreamStub = nil
+	fake.streamReturns = struct {
+		result1 string
+	}{result1}
+}
+
+func (fake *FakeMediaTrack) StreamReturnsOnCall(i int, result1 string) {
+	fake.streamMutex.Lock()
+	defer fake.streamMutex.Unlock()
+	fake.StreamStub = nil
+	if fake.streamReturnsOnCall == nil {
+		fake.streamReturnsOnCall = make(map[int]struct {
+			result1 string
+		})
+	}
+	fake.streamReturnsOnCall[i] = struct {
+		result1 string
+	}{result1}
+}
+
 func (fake *FakeMediaTrack) ToProto() *livekit.TrackInfo {
 	fake.toProtoMutex.Lock()
 	ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)]
@@ -1615,6 +1678,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} {
 	defer fake.setMutedMutex.RUnlock()
 	fake.sourceMutex.RLock()
 	defer fake.sourceMutex.RUnlock()
+	fake.streamMutex.RLock()
+	defer fake.streamMutex.RUnlock()
 	fake.toProtoMutex.RLock()
 	defer fake.toProtoMutex.RUnlock()
 	fake.updateVideoLayersMutex.RLock()

pkg/rtc/utils.go

@@ -42,6 +42,25 @@ func PackStreamID(participantID livekit.ParticipantID, trackID livekit.TrackID)
 	return string(participantID) + trackIdSeparator + string(trackID)
 }
 
+func PackSyncStreamID(participantID livekit.ParticipantID, stream string) string {
+	return string(participantID) + trackIdSeparator + stream
+}
+
+func StreamFromTrackSource(source livekit.TrackSource) string {
+	// group camera/mic, screenshare/audio together
+	switch source {
+	case livekit.TrackSource_SCREEN_SHARE:
+		return "screen"
+	case livekit.TrackSource_SCREEN_SHARE_AUDIO:
+		return "screen"
+	case livekit.TrackSource_CAMERA:
+		return "camera"
+	case livekit.TrackSource_MICROPHONE:
+		return "camera"
+	}
+	return "unknown"
+}
+
 func PackDataTrackLabel(participantID livekit.ParticipantID, trackID livekit.TrackID, label string) string {
 	return string(participantID) + trackIdSeparator + string(trackID) + trackIdSeparator + label
 }

pkg/service/roomallocator.go

@@ -151,4 +151,9 @@ func applyDefaultRoomConfig(room *livekit.Room, conf *config.RoomConfig) {
 			FmtpLine: codec.FmtpLine,
 		})
 	}
+	room.PlayoutDelay = &livekit.PlayoutDelay{
+		Enabled: conf.PlayoutDelay.Enabled,
+		Min:     uint32(conf.PlayoutDelay.Min),
+		Max:     uint32(conf.PlayoutDelay.Max),
+	}
 }

pkg/service/roommanager.go

@@ -390,6 +390,7 @@ func (r *RoomManager) StartSession(
 		SubscriberAllowPause:         subscriberAllowPause,
 		SubscriptionLimitAudio:       r.config.Limit.SubscriptionLimitAudio,
 		SubscriptionLimitVideo:       r.config.Limit.SubscriptionLimitVideo,
+		PlayoutDelay:                 protoRoom.PlayoutDelay,
 	})
 	if err != nil {
 		return err

pkg/sfu/buffer/buffer.go

@@ -181,7 +181,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 
 	for _, ext := range params.HeaderExtensions {
 		switch ext.URI {
-		case dd.ExtensionUrl:
+		case dd.ExtensionURI:
 			b.ddExt = uint8(ext.ID)
 			frc := NewFrameRateCalculatorDD(b.clockRate, b.logger)
 			for i := range b.frameRateCalculator {

pkg/sfu/dependencydescriptor/dependencydescriptorextension.go

@@ -69,7 +69,7 @@ const (
 
 	AllChainsAreActive = uint32(0)
 
-	ExtensionUrl = "https://aomediacodec.github.io/av1-rtp-spec/#dependency-descriptor-rtp-header-extension"
+	ExtensionURI = "https://aomediacodec.github.io/av1-rtp-spec/#dependency-descriptor-rtp-header-extension"
 )
 
 // ------------------------------------------------------------------------------

pkg/sfu/downtrack.go

@@ -37,6 +37,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
 	dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
 	"github.com/livekit/livekit-server/pkg/sfu/pacer"
+	"github.com/livekit/livekit-server/pkg/sfu/rtpextension"
 )
 
 // TrackSender defines an interface send media to remote peer
@@ -183,6 +184,19 @@ type DownTrackStreamAllocatorListener interface {
 
 type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport)
 
+type DowntrackParams struct {
+	Codecs            []webrtc.RTPCodecParameters
+	Receiver          TrackReceiver
+	BufferFactory     *buffer.Factory
+	SubID             livekit.ParticipantID
+	StreamID          string
+	MaxTrack          int
+	PlayoutDelayLimit *livekit.PlayoutDelay
+	Pacer             pacer.Pacer
+	Logger            logger.Logger
+	Trailer           []byte
+}
+
 // DownTrack implements TrackLocal, is the track used to write packets
 // to SFU Subscriber, the track handle the packets for simple, simulcast
 // and SVC Publisher.
@@ -192,17 +206,13 @@ type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport)
 // - closed
 // once closed, a DownTrack cannot be re-used.
 type DownTrack struct {
-	logger        logger.Logger
-	id            livekit.TrackID
-	subscriberID  livekit.ParticipantID
-	kind          webrtc.RTPCodecType
-	mime          string
-	ssrc          uint32
-	streamID      string
-	maxTrack      int
-	payloadType   uint8
-	sequencer     *sequencer
-	bufferFactory *buffer.Factory
+	params      DowntrackParams
+	id          livekit.TrackID
+	kind        webrtc.RTPCodecType
+	mime        string
+	ssrc        uint32
+	payloadType uint8
+	sequencer   *sequencer
 
 	forwarder *Forwarder
 
@@ -211,8 +221,8 @@ type DownTrack struct {
 	absSendTimeExtID          int
 	transportWideExtID        int
 	dependencyDescriptorExtID int
-	receiver                  TrackReceiver
-	transceiver               *webrtc.RTPTransceiver
+	playoutDelayExtID         int
+	transceiver               atomic.Pointer[webrtc.RTPTransceiver]
 	writeStream               webrtc.TrackLocalWriter
 	rtcpReader                *buffer.RTCPReader
 
@@ -250,12 +260,13 @@ type DownTrack struct {
 	bytesSent                       atomic.Uint32
 	bytesRetransmitted              atomic.Uint32
 
+	playoutDelayBytes atomic.Value //bytes of marshalled playout delay
+	playoudDelayAcked atomic.Bool
+
 	pacer pacer.Pacer
 
 	maxLayerNotifierCh chan struct{}
 
-	trailer []byte
-
 	cbMu                        sync.RWMutex
 	onStatsUpdate               func(dt *DownTrack, stat *livekit.AnalyticsStat)
 	onMaxSubscribedLayerChanged func(dt *DownTrack, layer int32)
@@ -264,16 +275,8 @@ type DownTrack struct {
 }
 
 // NewDownTrack returns a DownTrack.
-func NewDownTrack(
-	codecs []webrtc.RTPCodecParameters,
-	r TrackReceiver,
-	bf *buffer.Factory,
-	subID livekit.ParticipantID,
-	mt int,
-	pacer pacer.Pacer,
-	trailer []byte,
-	logger logger.Logger,
-) (*DownTrack, error) {
+func NewDownTrack(params DowntrackParams) (*DownTrack, error) {
+	codecs := params.Codecs
 	var kind webrtc.RTPCodecType
 	switch {
 	case strings.HasPrefix(codecs[0].MimeType, "audio/"):
@@ -285,24 +288,18 @@ func NewDownTrack(
 	}
 
 	d := &DownTrack{
-		logger:             logger,
-		id:                 r.TrackID(),
-		subscriberID:       subID,
-		maxTrack:           mt,
-		streamID:           r.StreamID(),
-		bufferFactory:      bf,
-		receiver:           r,
+		params:             params,
+		id:                 params.Receiver.TrackID(),
 		upstreamCodecs:     codecs,
 		kind:               kind,
 		codec:              codecs[0].RTPCodecCapability,
-		pacer:              pacer,
-		trailer:            trailer,
+		pacer:              params.Pacer,
 		maxLayerNotifierCh: make(chan struct{}, 20),
 	}
 	d.forwarder = NewForwarder(
 		d.kind,
-		d.logger,
-		d.receiver.GetReferenceLayerRTPTimestamp,
+		params.Logger,
+		d.params.Receiver.GetReferenceLayerRTPTimestamp,
 		d.getExpectedRTPTimestamp,
 	)
 	d.forwarder.OnParkedLayerExpired(func() {
@@ -314,7 +311,7 @@ func NewDownTrack(
 	d.rtpStats = buffer.NewRTPStats(buffer.RTPStatsParams{
 		ClockRate:              d.codec.ClockRate,
 		IsReceiverReportDriven: true,
-		Logger:                 d.logger,
+		Logger:                 params.Logger,
 	})
 	d.deltaStatsSnapshotId = d.rtpStats.NewSnapshotId()
 	d.deltaStatsOverriddenSnapshotId = d.rtpStats.NewSnapshotId()
@@ -325,7 +322,7 @@ func NewDownTrack(
 		GetDeltaStats:             d.getDeltaStats,
 		GetDeltaStatsOverridden:   d.getDeltaStatsOverridden,
 		GetLastReceiverReportTime: func() time.Time { return d.rtpStats.LastReceiverReport() },
-		Logger:                    d.logger.WithValues("direction", "down"),
+		Logger:                    params.Logger.WithValues("direction", "down"),
 	})
 	d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) {
 		if onStatsUpdate := d.getOnStatsUpdate(); onStatsUpdate != nil {
@@ -333,6 +330,18 @@ func NewDownTrack(
 		}
 	})
 
+	// set initial playout delay to minimum value
+
+	if d.params.PlayoutDelayLimit.GetEnabled() && d.params.PlayoutDelayLimit.GetMin() > 0 {
+		delay := rtpextension.PlayoutDelayFromValue(
+			uint16(d.params.PlayoutDelayLimit.GetMin()),
+			uint16(d.params.PlayoutDelayLimit.GetMax()),
+		)
+		b, err := delay.Marshal()
+		if err == nil {
+			d.playoutDelayBytes.Store(b)
+		}
+	}
 	if d.kind == webrtc.RTPCodecTypeVideo {
 		go d.maxLayerNotifierWorker()
 	}
@@ -362,7 +371,7 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
 		err := webrtc.ErrUnsupportedCodec
 		onBinding := d.onBinding
 		d.bindLock.Unlock()
-		d.logger.Infow("bind error for unsupported codec", "codecs", d.upstreamCodecs, "remoteParameters", t.CodecParameters())
+		d.params.Logger.Infow("bind error for unsupported codec", "codecs", d.upstreamCodecs, "remoteParameters", t.CodecParameters())
 		if onBinding != nil {
 			onBinding(err)
 		}
@@ -371,17 +380,17 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
 
 	// if a downtrack is closed before bind, it already unsubscribed from client, don't do subsequent operation and return here.
 	if d.IsClosed() {
-		d.logger.Debugw("DownTrack closed before bind")
+		d.params.Logger.Debugw("DownTrack closed before bind")
 		d.bindLock.Unlock()
 		return codec, nil
 	}
 
-	d.logger.Debugw("DownTrack.Bind", "codecs", d.upstreamCodecs, "matchCodec", codec, "ssrc", t.SSRC())
+	d.params.Logger.Debugw("DownTrack.Bind", "codecs", d.upstreamCodecs, "matchCodec", codec, "ssrc", t.SSRC())
 	d.ssrc = uint32(t.SSRC())
 	d.payloadType = uint8(codec.PayloadType)
 	d.writeStream = t.WriteStream()
 	d.mime = strings.ToLower(codec.MimeType)
-	if rr := d.bufferFactory.GetOrNew(packetio.RTCPBufferPacket, uint32(t.SSRC())).(*buffer.RTCPReader); rr != nil {
+	if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, uint32(t.SSRC())).(*buffer.RTCPReader); rr != nil {
 		rr.OnPacket(func(pkt []byte) {
 			d.handleRTCP(pkt)
 		})
@@ -389,9 +398,9 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
 	}
 
 	if d.kind == webrtc.RTPCodecTypeAudio {
-		d.sequencer = newSequencer(d.maxTrack, 0, d.logger)
+		d.sequencer = newSequencer(d.params.MaxTrack, 0, d.params.Logger)
 	} else {
-		d.sequencer = newSequencer(d.maxTrack, maxPadding, d.logger)
+		d.sequencer = newSequencer(d.params.MaxTrack, maxPadding, d.params.Logger)
 	}
 
 	d.codec = codec.RTPCodecCapability
@@ -401,9 +410,18 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
 	d.bound.Store(true)
 	d.bindLock.Unlock()
 
-	d.forwarder.DetermineCodec(d.codec, d.receiver.HeaderExtensions())
+	// Bind is called under RTPSender.mu lock, call the RTPSender.GetParameters in goroutine to avoid deadlock
+	go func() {
+		if tr := d.transceiver.Load(); tr != nil {
+			extensions := tr.Sender().GetParameters().HeaderExtensions
+			d.params.Logger.Debugw("negotiated downtrack extensions", "extensions", extensions)
+			d.SetRTPHeaderExtensions(extensions)
+		}
+	}()
 
-	d.logger.Debugw("downtrack bound")
+	d.forwarder.DetermineCodec(d.codec, d.params.Receiver.HeaderExtensions())
+
+	d.params.Logger.Debugw("downtrack bound")
 	d.onBindAndConnected()
 
 	return codec, nil
@@ -417,7 +435,7 @@ func (d *DownTrack) Unbind(_ webrtc.TrackLocalContext) error {
 }
 
 func (d *DownTrack) TrackInfoAvailable() {
-	ti := d.receiver.TrackInfo()
+	ti := d.params.Receiver.TrackInfo()
 	if ti == nil {
 		return
 	}
@@ -495,9 +513,9 @@ func (d *DownTrack) ID() string { return string(d.id) }
 func (d *DownTrack) Codec() webrtc.RTPCodecCapability { return d.codec }
 
 // StreamID is the group this track belongs too. This must be unique
-func (d *DownTrack) StreamID() string { return d.streamID }
+func (d *DownTrack) StreamID() string { return d.params.StreamID }
 
-func (d *DownTrack) SubscriberID() livekit.ParticipantID { return d.subscriberID }
+func (d *DownTrack) SubscriberID() livekit.ParticipantID { return d.params.SubID }
 
 // Sets RTP header extensions for this track
 func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeaderExtensionParameter) {
@@ -505,10 +523,12 @@ func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeade
 		switch ext.URI {
 		case sdp.ABSSendTimeURI:
 			d.absSendTimeExtID = ext.ID
+		case dd.ExtensionURI:
+			d.dependencyDescriptorExtID = ext.ID
+		case rtpextension.PlayoutDelayURI:
+			d.playoutDelayExtID = ext.ID
 		case sdp.TransportCCURI:
 			d.transportWideExtID = ext.ID
-		case dd.ExtensionUrl:
-			d.dependencyDescriptorExtID = ext.ID
 		}
 	}
 }
@@ -528,18 +548,18 @@ func (d *DownTrack) SSRC() uint32 {
 }
 
 func (d *DownTrack) Stop() error {
-	if d.transceiver != nil {
-		return d.transceiver.Stop()
+	if tr := d.transceiver.Load(); tr != nil {
+		return tr.Stop()
 	}
 	return errors.New("downtrack transceiver does not exist")
 }
 
 func (d *DownTrack) SetTransceiver(transceiver *webrtc.RTPTransceiver) {
-	d.transceiver = transceiver
+	d.transceiver.Store(transceiver)
 }
 
 func (d *DownTrack) GetTransceiver() *webrtc.RTPTransceiver {
-	return d.transceiver
+	return d.transceiver.Load()
 }
 
 func (d *DownTrack) maybeStartKeyFrameRequester() {
@@ -551,7 +571,6 @@ func (d *DownTrack) maybeStartKeyFrameRequester() {
 	//
 	d.stopKeyFrameRequester()
 
-	// SVC-TODO : don't need pli/lrr when layer comes down
 	locked, layer := d.forwarder.CheckSync()
 	if !locked {
 		go d.keyFrameRequester(d.keyFrameRequestGeneration.Load(), layer)
@@ -584,8 +603,8 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) {
 		}
 
 		if d.connected.Load() {
-			d.logger.Debugw("sending PLI for layer lock", "generation", generation, "layer", layer)
-			d.receiver.SendPLI(layer, false)
+			d.params.Logger.Debugw("sending PLI for layer lock", "generation", generation, "layer", layer)
+			d.params.Receiver.SendPLI(layer, false)
 			d.rtpStats.UpdateLayerLockPliAndTime(1)
 		}
 
@@ -605,7 +624,7 @@ func (d *DownTrack) postMaxLayerNotifierEvent() {
 	select {
 	case d.maxLayerNotifierCh <- struct{}{}:
 	default:
-		d.logger.Warnw("max layer notifier event queue full", nil)
+		d.params.Logger.Warnw("max layer notifier event queue full", nil)
 	}
 }
 
@@ -619,7 +638,7 @@ func (d *DownTrack) maxLayerNotifierWorker() {
 			maxLayerSpatial = d.forwarder.GetMaxSubscribedSpatial()
 		}
 		if onMaxSubscribedLayerChanged := d.getOnMaxLayerChanged(); onMaxSubscribedLayerChanged != nil {
-			d.logger.Infow("max subscribed layer changed", "maxLayerSpatial", maxLayerSpatial)
+			d.params.Logger.Infow("max subscribed layer changed", "maxLayerSpatial", maxLayerSpatial)
 			onMaxSubscribedLayerChanged(d, maxLayerSpatial)
 		}
 	}
@@ -634,7 +653,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 	tp, err := d.forwarder.GetTranslationParams(extPkt, layer)
 	if tp.shouldDrop {
 		if err != nil {
-			d.logger.Errorw("write rtp packet failed", err)
+			d.params.Logger.Errorw("write rtp packet failed", err)
 		}
 		return err
 	}
@@ -654,13 +673,19 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 
 	hdr, err := d.getTranslatedRTPHeader(extPkt, tp)
 	if err != nil {
-		d.logger.Errorw("write rtp packet failed", err)
+		d.params.Logger.Errorw("write rtp packet failed", err)
 		if pool != nil {
 			PacketFactory.Put(pool)
 		}
 		return err
 	}
 
+	extensions := []pacer.ExtensionData{{ID: uint8(d.dependencyDescriptorExtID), Payload: tp.ddBytes}}
+	if d.playoutDelayExtID != 0 && !d.playoudDelayAcked.Load() {
+		if val := d.playoutDelayBytes.Load(); val != nil {
+			extensions = append(extensions, pacer.ExtensionData{ID: uint8(d.playoutDelayExtID), Payload: val.([]byte)})
+		}
+	}
 	if d.sequencer != nil {
 		d.sequencer.push(
 			extPkt.Packet.SequenceNumber,
@@ -675,7 +700,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 
 	d.pacer.Enqueue(pacer.Packet{
 		Header:             hdr,
-		Extensions:         []pacer.ExtensionData{{ID: uint8(d.dependencyDescriptorExtID), Payload: tp.ddBytes}},
+		Extensions:         extensions,
 		Payload:            payload,
 		AbsSendTimeExtID:   uint8(d.absSendTimeExtID),
 		TransportWideExtID: uint8(d.transportWideExtID),
@@ -869,7 +894,7 @@ func (d *DownTrack) CloseWithFlush(flush bool) {
 	}
 
 	d.bindLock.Lock()
-	d.logger.Debugw("close down track", "flushBlankFrame", flush)
+	d.params.Logger.Debugw("close down track", "flushBlankFrame", flush)
 	if d.bound.Load() {
 		if d.forwarder != nil {
 			d.forwarder.Mute(true)
@@ -893,12 +918,12 @@ func (d *DownTrack) CloseWithFlush(flush bool) {
 		}
 
 		d.bound.Store(false)
-		d.logger.Debugw("closing sender", "kind", d.kind)
+		d.params.Logger.Debugw("closing sender", "kind", d.kind)
 	}
-	d.receiver.DeleteDownTrack(d.subscriberID)
+	d.params.Receiver.DeleteDownTrack(d.params.SubID)
 
 	if d.rtcpReader != nil && flush {
-		d.logger.Debugw("downtrack close rtcp reader")
+		d.params.Logger.Debugw("downtrack close rtcp reader")
 		d.rtcpReader.Close()
 		d.rtcpReader.OnPacket(nil)
 	}
@@ -906,7 +931,7 @@ func (d *DownTrack) CloseWithFlush(flush bool) {
 	d.bindLock.Unlock()
 	d.connectionStats.Close()
 	d.rtpStats.Stop()
-	d.logger.Infow("rtp stats", "direction", "downstream", "mime", d.mime, "ssrc", d.ssrc, "stats", d.rtpStats.ToString())
+	d.params.Logger.Infow("rtp stats", "direction", "downstream", "mime", d.mime, "ssrc", d.ssrc, "stats", d.rtpStats.ToString())
 
 	close(d.maxLayerNotifierCh)
 
@@ -1088,17 +1113,17 @@ func (d *DownTrack) IsDeficient() bool {
 }
 
 func (d *DownTrack) BandwidthRequested() int64 {
-	_, brs := d.receiver.GetLayeredBitrate()
+	_, brs := d.params.Receiver.GetLayeredBitrate()
 	return d.forwarder.BandwidthRequested(brs)
 }
 
 func (d *DownTrack) DistanceToDesired() float64 {
-	al, brs := d.receiver.GetLayeredBitrate()
+	al, brs := d.params.Receiver.GetLayeredBitrate()
 	return d.forwarder.DistanceToDesired(al, brs)
 }
 
 func (d *DownTrack) AllocateOptimal(allowOvershoot bool) VideoAllocation {
-	al, brs := d.receiver.GetLayeredBitrate()
+	al, brs := d.params.Receiver.GetLayeredBitrate()
 	allocation := d.forwarder.AllocateOptimal(al, brs, allowOvershoot)
 	d.maybeStartKeyFrameRequester()
 	d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason)
@@ -1106,7 +1131,7 @@ func (d *DownTrack) AllocateOptimal(allowOvershoot bool) VideoAllocation {
 }
 
 func (d *DownTrack) ProvisionalAllocatePrepare() {
-	al, brs := d.receiver.GetLayeredBitrate()
+	al, brs := d.params.Receiver.GetLayeredBitrate()
 	d.forwarder.ProvisionalAllocatePrepare(al, brs)
 }
 
@@ -1120,13 +1145,13 @@ func (d *DownTrack) ProvisionalAllocate(availableChannelCapacity int64, layers b
 
 func (d *DownTrack) ProvisionalAllocateGetCooperativeTransition(allowOvershoot bool) VideoTransition {
 	transition := d.forwarder.ProvisionalAllocateGetCooperativeTransition(allowOvershoot)
-	d.logger.Debugw("stream: cooperative transition", "transition", transition)
+	d.params.Logger.Debugw("stream: cooperative transition", "transition", transition)
 	return transition
 }
 
 func (d *DownTrack) ProvisionalAllocateGetBestWeightedTransition() VideoTransition {
 	transition := d.forwarder.ProvisionalAllocateGetBestWeightedTransition()
-	d.logger.Debugw("stream: best weighted transition", "transition", transition)
+	d.params.Logger.Debugw("stream: best weighted transition", "transition", transition)
 	return transition
 }
 
@@ -1138,7 +1163,7 @@ func (d *DownTrack) ProvisionalAllocateCommit() VideoAllocation {
 }
 
 func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64, allowOvershoot bool) (VideoAllocation, bool) {
-	al, brs := d.receiver.GetLayeredBitrate()
+	al, brs := d.params.Receiver.GetLayeredBitrate()
 	allocation, available := d.forwarder.AllocateNextHigher(availableChannelCapacity, al, brs, allowOvershoot)
 	d.maybeStartKeyFrameRequester()
 	d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason)
@@ -1146,14 +1171,14 @@ func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64, allowOver
 }
 
 func (d *DownTrack) GetNextHigherTransition(allowOvershoot bool) (VideoTransition, bool) {
-	_, brs := d.receiver.GetLayeredBitrate()
+	_, brs := d.params.Receiver.GetLayeredBitrate()
 	transition, available := d.forwarder.GetNextHigherTransition(brs, allowOvershoot)
-	d.logger.Debugw("stream: get next higher layer", "transition", transition, "available", available, "bitrates", brs)
+	d.params.Logger.Debugw("stream: get next higher layer", "transition", transition, "available", available, "bitrates", brs)
 	return transition, available
 }
 
 func (d *DownTrack) Pause() VideoAllocation {
-	al, brs := d.receiver.GetLayeredBitrate()
+	al, brs := d.params.Receiver.GetLayeredBitrate()
 	allocation := d.forwarder.Pause(al, brs)
 	d.maybeStartKeyFrameRequester()
 	d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason)
@@ -1165,7 +1190,7 @@ func (d *DownTrack) Resync() {
 }
 
 func (d *DownTrack) CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChunk {
-	if !d.bound.Load() {
+	if !d.bound.Load() || d.transceiver.Load() == nil {
 		return nil
 	}
 	return []rtcp.SourceDescriptionChunk{
@@ -1173,13 +1198,13 @@ func (d *DownTrack) CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChun
 			Source: d.ssrc,
 			Items: []rtcp.SourceDescriptionItem{{
 				Type: rtcp.SDESCNAME,
-				Text: d.streamID,
+				Text: d.params.StreamID,
 			}},
 		}, {
 			Source: d.ssrc,
 			Items: []rtcp.SourceDescriptionItem{{
 				Type: rtcp.SDESType(15),
-				Text: d.transceiver.Mid(),
+				Text: d.transceiver.Load().Mid(),
 			}},
 		},
 	}
@@ -1194,7 +1219,7 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport {
 	if clockLayer == buffer.InvalidLayerSpatial {
 		clockLayer = d.forwarder.GetReferenceLayerSpatial()
 	}
-	return d.rtpStats.GetRtcpSenderReport(d.ssrc, d.receiver.GetCalculatedClockRate(clockLayer))
+	return d.rtpStats.GetRtcpSenderReport(d.ssrc, d.params.Receiver.GetCalculatedClockRate(clockLayer))
 }
 
 func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan struct{} {
@@ -1243,7 +1268,7 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan
 
 			snts, frameEndNeeded, err := d.forwarder.GetSnTsForBlankFrames(frameRate, 1)
 			if err != nil {
-				d.logger.Warnw("could not get SN/TS for blank frame", err)
+				d.params.Logger.Warnw("could not get SN/TS for blank frame", err)
 				close(done)
 				return
 			}
@@ -1262,7 +1287,7 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan
 
 				payload, err := getBlankFrame(frameEndNeeded)
 				if err != nil {
-					d.logger.Warnw("could not get blank frame", err)
+					d.params.Logger.Warnw("could not get blank frame", err)
 					close(done)
 					return
 				}
@@ -1293,13 +1318,13 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan
 }
 
 func (d *DownTrack) maybeAddTrailer(buf []byte) int {
-	if len(buf) < len(d.trailer) {
-		d.logger.Warnw("trailer too big", nil, "bufLen", len(buf), "trailerLen", len(d.trailer))
+	if len(buf) < len(d.params.Trailer) {
+		d.params.Logger.Warnw("trailer too big", nil, "bufLen", len(buf), "trailerLen", len(d.params.Trailer))
 		return 0
 	}
 
-	copy(buf, d.trailer)
-	return len(d.trailer)
+	copy(buf, d.params.Trailer)
+	return len(d.params.Trailer)
 }
 
 func (d *DownTrack) getOpusBlankFrame(_frameEndNeeded bool) ([]byte, error) {
@@ -1366,7 +1391,7 @@ func (d *DownTrack) getH264BlankFrame(_frameEndNeeded bool) ([]byte, error) {
 func (d *DownTrack) handleRTCP(bytes []byte) {
 	pkts, err := rtcp.Unmarshal(bytes)
 	if err != nil {
-		d.logger.Errorw("unmarshal rtcp receiver packets err", err)
+		d.params.Logger.Errorw("unmarshal rtcp receiver packets err", err)
 		return
 	}
 
@@ -1375,8 +1400,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 		if pliOnce {
 			_, layer := d.forwarder.CheckSync()
 			if layer != buffer.InvalidLayerSpatial && !d.forwarder.IsAnyMuted() {
-				d.logger.Debugw("sending PLI RTCP", "layer", layer)
-				d.receiver.SendPLI(layer, false)
+				d.params.Logger.Debugw("sending PLI RTCP", "layer", layer)
+				d.params.Receiver.SendPLI(layer, false)
 				d.isNACKThrottled.Store(true)
 				d.rtpStats.UpdatePliTime()
 				pliOnce = false
@@ -1424,6 +1449,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 				if sal := d.getStreamAllocatorListener(); sal != nil {
 					sal.OnRTCPReceiverReport(d, r)
 				}
+
+				d.playoudDelayAcked.Store(true)
 			}
 			if len(rr.Reports) > 0 {
 				d.listenerLock.RLock()
@@ -1513,7 +1540,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 		})
 
 		pktBuff := *src
-		n, err := d.receiver.ReadRTP(pktBuff, uint8(meta.layer), meta.sourceSeqNo)
+		n, err := d.params.Receiver.ReadRTP(pktBuff, uint8(meta.layer), meta.sourceSeqNo)
 		if err != nil {
 			if err == io.EOF {
 				break
@@ -1528,7 +1555,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 
 		var pkt rtp.Packet
 		if err = pkt.Unmarshal(pktBuff[:n]); err != nil {
-			d.logger.Errorw("unmarshalling rtp packet failed in retransmit", err)
+			d.params.Logger.Errorw("unmarshalling rtp packet failed in retransmit", err)
 			continue
 		}
 		pkt.Header.Marker = meta.marker
@@ -1542,7 +1569,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 		if d.mime == "video/vp8" && len(pkt.Payload) > 0 && len(meta.codecBytes) != 0 {
 			var incomingVP8 buffer.VP8
 			if err = incomingVP8.Unmarshal(pkt.Payload); err != nil {
-				d.logger.Errorw("unmarshalling VP8 packet err", err)
+				d.params.Logger.Errorw("unmarshalling VP8 packet err", err)
 				PacketFactory.Put(pool)
 				continue
 			}
@@ -1633,9 +1660,9 @@ func (d *DownTrack) DebugInfo() map[string]interface{} {
 	}
 
 	return map[string]interface{}{
-		"SubscriberID":        d.subscriberID,
+		"SubscriberID":        d.params.SubID,
 		"TrackID":             d.id,
-		"StreamID":            d.streamID,
+		"StreamID":            d.params.StreamID,
 		"SSRC":                d.ssrc,
 		"MimeType":            d.codec.MimeType,
 		"Bound":               d.bound.Load(),
@@ -1697,7 +1724,7 @@ func (d *DownTrack) onBindAndConnected() {
 		if d.kind == webrtc.RTPCodecTypeVideo {
 			_, layer := d.forwarder.CheckSync()
 			if layer != buffer.InvalidLayerSpatial {
-				d.receiver.SendPLI(layer, true)
+				d.params.Receiver.SendPLI(layer, true)
 			}
 		}
 
@@ -1711,7 +1738,7 @@ func (d *DownTrack) sendPaddingOnMute() {
 	// let uptrack have chance to send packet before we send padding
 	time.Sleep(waitBeforeSendPaddingOnMute)
 
-	d.logger.Debugw("sending padding on mute")
+	d.params.Logger.Debugw("sending padding on mute")
 	if d.kind == webrtc.RTPCodecTypeVideo {
 		d.sendPaddingOnMuteForVideo()
 	} else if d.mime == "audio/opus" {
@@ -1741,7 +1768,7 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() {
 		}
 		snts, _, err := d.forwarder.GetSnTsForBlankFrames(frameRate, 1)
 		if err != nil {
-			d.logger.Warnw("could not get SN/TS for blank frame", err)
+			d.params.Logger.Warnw("could not get SN/TS for blank frame", err)
 			return
 		}
 		for i := 0; i < len(snts); i++ {
@@ -1758,7 +1785,7 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() {
 
 			payload, err := d.getOpusBlankFrame(false)
 			if err != nil {
-				d.logger.Warnw("could not get blank frame", err)
+				d.params.Logger.Warnw("could not get blank frame", err)
 				return
 			}
 
@@ -1801,7 +1828,7 @@ type sendPacketMetadata struct {
 func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int, sendTime time.Time, sendError error) {
 	spmd, ok := md.(sendPacketMetadata)
 	if !ok {
-		d.logger.Errorw("invalid send packet metadata", nil)
+		d.params.Logger.Errorw("invalid send packet metadata", nil)
 		return
 	}
 
@@ -1840,7 +1867,7 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int,
 	if spmd.isKeyFrame {
 		d.isNACKThrottled.Store(false)
 		d.rtpStats.UpdateKeyFrame(1)
-		d.logger.Debugw(
+		d.params.Logger.Debugw(
 			"forwarded key frame",
 			"layer", spmd.layer,
 			"rtpsn", hdr.SequenceNumber,

pkg/sfu/forwarder.go

@@ -309,7 +309,7 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [
 	searchDone:
 		for _, ext := range extensions {
 			switch ext.URI {
-			case dd.ExtensionUrl:
+			case dd.ExtensionURI:
 				isDDAvailable = true
 				break searchDone
 			}

pkg/sfu/receiver.go

@@ -126,8 +126,8 @@ type WebRTCReceiver struct {
 	onStatsUpdate    func(w *WebRTCReceiver, stat *livekit.AnalyticsStat)
 	onMaxLayerChange func(maxLayer int32)
 
-	primaryReceiver atomic.Value // *RedPrimaryReceiver
-	redReceiver     atomic.Value // *RedReceiver
+	primaryReceiver atomic.Pointer[RedPrimaryReceiver]
+	redReceiver     atomic.Pointer[RedReceiver]
 	redPktWriter    func(pkt *buffer.ExtPacket, spatialLayer int32)
 }
 
@@ -232,7 +232,7 @@ func NewWebRTCReceiver(
 	w.connectionStats.Start(w.trackInfo)
 
 	for _, ext := range receiver.GetParameters().HeaderExtensions {
-		if ext.URI == dd.ExtensionUrl {
+		if ext.URI == dd.ExtensionURI {
 			w.streamTrackerManager.AddDependencyDescriptorTrackers()
 			break
 		}
@@ -624,10 +624,10 @@ func (w *WebRTCReceiver) forwardRTP(layer int32) {
 			w.closed.Store(true)
 			w.closeTracks()
 			if pr := w.primaryReceiver.Load(); pr != nil {
-				pr.(*RedPrimaryReceiver).Close()
+				pr.Close()
 			}
 			if pr := w.redReceiver.Load(); pr != nil {
-				pr.(*RedReceiver).Close()
+				pr.Close()
 			}
 		})
 
@@ -731,7 +731,7 @@ func (w *WebRTCReceiver) GetPrimaryReceiverForRed() TrackReceiver {
 			w.bufferMu.Unlock()
 		}
 	}
-	return w.primaryReceiver.Load().(*RedPrimaryReceiver)
+	return w.primaryReceiver.Load()
 }
 
 func (w *WebRTCReceiver) GetRedReceiver() TrackReceiver {
@@ -750,7 +750,7 @@ func (w *WebRTCReceiver) GetRedReceiver() TrackReceiver {
 			w.bufferMu.Unlock()
 		}
 	}
-	return w.redReceiver.Load().(*RedReceiver)
+	return w.redReceiver.Load()
 }
 
 func (w *WebRTCReceiver) GetTemporalLayerFpsForSpatial(layer int32) []float32 {

pkg/sfu/rtpextension/playoutdelay.go

@@ -0,0 +1,56 @@
+package rtpextension
+
+import (
+	"encoding/binary"
+	"errors"
+)
+
+const (
+	PlayoutDelayURI = "http://www.webrtc.org/experiments/rtp-hdrext/playout-delay"
+
+	playoutDelayExtensionSize = 3
+)
+
+var (
+	errPlayoutDelayOverflow = errors.New("playout delay overflow")
+	errTooSmall             = errors.New("buffer too small")
+)
+
+//  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |  ID   | len=2 |       MIN delay       |       MAX delay       |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// The wired MIN/MAX delay is in 10ms unit
+
+type PlayOutDelay struct {
+	Min, Max uint16 // delay in ms
+}
+
+func PlayoutDelayFromValue(min, max uint16) PlayOutDelay {
+	if min >= (1<<12)*10 {
+		min = (1<<12 - 1) * 10
+	}
+	if max >= (1<<12)*10 {
+		max = (1<<12 - 1) * 10
+	}
+	return PlayOutDelay{Min: min, Max: max}
+}
+
+func (p PlayOutDelay) Marshal() ([]byte, error) {
+	min, max := p.Min/10, p.Max/10
+	if min >= 1<<12 || max >= 1<<12 {
+		return nil, errPlayoutDelayOverflow
+	}
+
+	return []byte{byte(min >> 4), byte(min<<4) | byte(max>>8), byte(max)}, nil
+}
+
+func (p *PlayOutDelay) Unmarshal(rawData []byte) error {
+	if len(rawData) < playoutDelayExtensionSize {
+		return errTooSmall
+	}
+
+	p.Min = (binary.BigEndian.Uint16(rawData) >> 4) * 10
+	p.Max = (binary.BigEndian.Uint16(rawData[1:]) & 0x0FFF) * 10
+	return nil
+}

pkg/sfu/rtpextension/playoutdelay_test.go

@@ -0,0 +1,35 @@
+package rtpextension
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestPlayoutDelay(t *testing.T) {
+	p1 := PlayOutDelay{Min: 100, Max: 200}
+	b, err := p1.Marshal()
+	require.NoError(t, err)
+	require.Len(t, b, playoutDelayExtensionSize)
+	var p2 PlayOutDelay
+	err = p2.Unmarshal(b)
+	require.NoError(t, err)
+	require.Equal(t, p1, p2)
+
+	// overflow
+	p3 := PlayOutDelay{Min: 100, Max: (1 << 12) * 10}
+	_, err = p3.Marshal()
+	require.ErrorIs(t, err, errPlayoutDelayOverflow)
+
+	// too small
+	p4 := PlayOutDelay{}
+	err = p4.Unmarshal([]byte{0x00, 0x00})
+	require.ErrorIs(t, err, errTooSmall)
+
+	// from value
+	p5 := PlayoutDelayFromValue(1<<12*10, 1<<12*10+10)
+	_, err = p5.Marshal()
+	require.NoError(t, err)
+	require.Equal(t, uint16((1<<12)-1)*10, p5.Min)
+	require.Equal(t, uint16((1<<12)-1)*10, p5.Max)
+}