Merge remote-tracking branch 'origin/master' into raja_min_packets

go.mod

@@ -17,9 +17,9 @@ require (
 	github.com/hashicorp/golang-lru/v2 v2.0.4
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
-	github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135
-	github.com/livekit/protocol v1.5.9
-	github.com/livekit/psrpc v0.3.1
+	github.com/livekit/mediatransportutil v0.0.0-20230716190407-fc4944cbc33a
+	github.com/livekit/protocol v1.5.10
+	github.com/livekit/psrpc v0.3.2
 	github.com/mackerelio/go-osstat v0.2.4
 	github.com/magefile/mage v1.15.0
 	github.com/maxbrunsfeld/counterfeiter/v6 v6.6.2
@@ -101,6 +101,6 @@ require (
 	golang.org/x/text v0.10.0 // indirect
 	golang.org/x/tools v0.9.3 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230530153820-e85fd2cbaebc // indirect
-	google.golang.org/grpc v1.56.1 // indirect
+	google.golang.org/grpc v1.56.2 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )

go.sum

@@ -122,12 +122,12 @@ github.com/lithammer/shortuuid/v4 v4.0.0 h1:QRbbVkfgNippHOS8PXDkti4NaWeyYfcBTHtw
 github.com/lithammer/shortuuid/v4 v4.0.0/go.mod h1:Zs8puNcrvf2rV9rTH51ZLLcj7ZXqQI3lv67aw4KiB1Y=
 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkDaKb5iXdynYrzB84ErPPO4LbRASk58=
 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
-github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135 h1:lWYbsondvqG69czxoACDwaJ/BoyD57BahCo70ZH+m4U=
-github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135/go.mod h1:MRc0zSOSzXuFt0X218SgabzlaKevkvCckPgBEoHYc34=
-github.com/livekit/protocol v1.5.9 h1:fqPOLgKkWmkmUMnpfj2KDZlidHgAazAPU2T3FewyLew=
-github.com/livekit/protocol v1.5.9/go.mod h1:GMTlFbc0JypUGo+PMilDrL45AX+yUBiQ1Cl/FGZibDY=
-github.com/livekit/psrpc v0.3.1 h1:KfylgJHvoLQcc22t/oflwMOeSnx0c14G7cWsS+9MYS4=
-github.com/livekit/psrpc v0.3.1/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U=
+github.com/livekit/mediatransportutil v0.0.0-20230716190407-fc4944cbc33a h1:JWpPHcMFuw0fP4swE89CfMgeUXiSN5IKvCJL/5HLI3A=
+github.com/livekit/mediatransportutil v0.0.0-20230716190407-fc4944cbc33a/go.mod h1:xirUXW8xnLGmfCwUeAv/nj1VGo1OO1BmgxrYP7jK/14=
+github.com/livekit/protocol v1.5.10 h1:lnaHMa27cbRkHybi/jvOVuRSaLsho2wCLRjKiC6ce2Y=
+github.com/livekit/protocol v1.5.10/go.mod h1:eRzojAYSPJuNgDHMlvLji/CPauj9hrgvb6rVPUj6MoU=
+github.com/livekit/psrpc v0.3.2 h1:eAaJhASme33gtoBhCRLH9jsnWcdm1tHWf0WzaDk56ew=
+github.com/livekit/psrpc v0.3.2/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U=
 github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs=
 github.com/mackerelio/go-osstat v0.2.4/go.mod h1:Zy+qzGdZs3A9cuIqmgbJvwbmLQH9dJvtio5ZjJTbdlQ=
 github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=
@@ -411,8 +411,8 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20230530153820-e85fd2cbaebc h1:XSJ8Vk1SWuNr8S18z1NZSziL0CPIXLCCMDOEFtHBOFc=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20230530153820-e85fd2cbaebc/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA=
-google.golang.org/grpc v1.56.1 h1:z0dNfjIl0VpaZ9iSVjA6daGatAYwPGstTjt5vkRMFkQ=
-google.golang.org/grpc v1.56.1/go.mod h1:I9bI3vqKfayGqPUAwGdOSu7kt6oIJLixfffKrpXqQ9s=
+google.golang.org/grpc v1.56.2 h1:fVRFRnXvU+x6C4IlHZewvJOVHoOv1TUuQyoRsYnB4bI=
+google.golang.org/grpc v1.56.2/go.mod h1:I9bI3vqKfayGqPUAwGdOSu7kt6oIJLixfffKrpXqQ9s=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=

pkg/rtc/participant.go

@@ -370,13 +370,14 @@ func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermissio
 		return false
 	}
 
-	p.GetLogger().Infow("updating participant permission", "permission", permission)
+	p.params.Logger.Infow("updating participant permission", "permission", permission)
 
 	video.UpdateFromPermission(permission)
 	p.dirty.Store(true)
 
 	canPublish := video.GetCanPublish()
 	canSubscribe := video.GetCanSubscribe()
+
 	onParticipantUpdate := p.onParticipantUpdate
 	onClaimsChanged := p.onClaimsChanged
 
@@ -387,13 +388,7 @@ func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermissio
 	// publish permission has been revoked then remove offending tracks
 	for _, track := range p.GetPublishedTracks() {
 		if !video.GetCanPublishSource(track.Source()) {
-			p.RemovePublishedTrack(track, false, false)
-			if p.ProtocolVersion().SupportsUnpublish() {
-				p.sendTrackUnpublished(track.ID())
-			} else {
-				// for older clients that don't support unpublish, mute to avoid them sending data
-				p.sendTrackMuted(track.ID(), true)
-			}
+			p.removePublishedTrack(track)
 		}
 	}
 
@@ -1190,22 +1185,24 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) {
 }
 
 func (p *ParticipantImpl) setIsPublisher(isPublisher bool) {
-	if p.isPublisher.Swap(isPublisher) != isPublisher {
-		p.lock.Lock()
-		p.requireBroadcast = true
-		p.lock.Unlock()
+	if p.isPublisher.Swap(isPublisher) == isPublisher {
+		return
+	}
 
-		p.dirty.Store(true)
+	p.lock.Lock()
+	p.requireBroadcast = true
+	p.lock.Unlock()
 
-		// trigger update as well if participant is already fully connected
-		if p.State() == livekit.ParticipantInfo_ACTIVE {
-			p.lock.RLock()
-			onParticipantUpdate := p.onParticipantUpdate
-			p.lock.RUnlock()
+	p.dirty.Store(true)
 
-			if onParticipantUpdate != nil {
-				onParticipantUpdate(p)
-			}
+	// trigger update as well if participant is already fully connected
+	if p.State() == livekit.ParticipantInfo_ACTIVE {
+		p.lock.RLock()
+		onParticipantUpdate := p.onParticipantUpdate
+		p.lock.RUnlock()
+
+		if onParticipantUpdate != nil {
+			onParticipantUpdate(p)
 		}
 	}
 }
@@ -1220,6 +1217,16 @@ func (p *ParticipantImpl) onSubscriberOffer(offer webrtc.SessionDescription) err
 	})
 }
 
+func (p *ParticipantImpl) removePublishedTrack(track types.MediaTrack) {
+	p.RemovePublishedTrack(track, false, false)
+	if p.ProtocolVersion().SupportsUnpublish() {
+		p.sendTrackUnpublished(track.ID())
+	} else {
+		// for older clients that don't support unpublish, mute to avoid them sending data
+		p.sendTrackMuted(track.ID(), true)
+	}
+}
+
 // when a new remoteTrack is created, creates a Track and adds it to room
 func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
 	if p.IsDisconnected() {
@@ -1242,12 +1249,12 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
 		p.params.Logger.Warnw("no permission to publish mediaTrack", nil,
 			"source", publishedTrack.Source(),
 		)
+		p.removePublishedTrack(publishedTrack)
 		return
 	}
 
-	if !p.IsPublisher() {
-		p.setIsPublisher(true)
-	}
+	p.setIsPublisher(true)
+	p.dirty.Store(true)
 
 	p.params.Logger.Infow("mediaTrack published",
 		"kind", track.Kind().String(),
@@ -1258,8 +1265,6 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
 		"mime", track.Codec().MimeType,
 	)
 
-	p.dirty.Store(true)
-
 	if !isNewTrack && !publishedTrack.HasPendingCodec() && p.IsReady() {
 		p.lock.RLock()
 		onTrackUpdated := p.onTrackUpdated
@@ -1300,9 +1305,7 @@ func (p *ParticipantImpl) onDataMessage(kind livekit.DataPacket_Kind, data []byt
 		p.params.Logger.Warnw("received unsupported data packet", nil, "payload", payload)
 	}
 
-	if !p.IsPublisher() {
-		p.setIsPublisher(true)
-	}
+	p.setIsPublisher(true)
 }
 
 func (p *ParticipantImpl) onICECandidate(c *webrtc.ICECandidate, target livekit.SignalTarget) error {

pkg/rtc/room.go

@@ -896,6 +896,7 @@ func (r *Room) onTrackUnpublished(p types.LocalParticipant, track types.MediaTra
 }
 
 func (r *Room) onParticipantUpdate(p types.LocalParticipant) {
+	r.protoProxy.MarkDirty(false)
 	// immediately notify when permissions or metadata changed
 	r.broadcastParticipantState(p, broadcastOptions{immediate: true})
 	if r.onParticipantChanged != nil {

pkg/rtc/wrappedreceiver.go

@@ -12,7 +12,6 @@ import (
 	"github.com/livekit/protocol/logger"
 
 	"github.com/livekit/livekit-server/pkg/sfu"
-	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 )
 
 // wrapper around WebRTC receiver, overriding its ID
@@ -297,11 +296,11 @@ func (d *DummyReceiver) GetRedReceiver() sfu.TrackReceiver {
 	return d
 }
 
-func (d *DummyReceiver) GetRTCPSenderReportData(layer int32) (*buffer.RTCPSenderReportData, *buffer.RTCPSenderReportData) {
+func (d *DummyReceiver) GetCalculatedClockRate(layer int32) uint32 {
 	if r, ok := d.receiver.Load().(sfu.TrackReceiver); ok {
-		return r.GetRTCPSenderReportData(layer)
+		return r.GetCalculatedClockRate(layer)
 	}
-	return nil, nil
+	return 0
 }
 
 func (d *DummyReceiver) GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) {

pkg/service/ingress.go

@@ -216,10 +216,19 @@ func (s *IngressService) ListIngress(ctx context.Context, req *livekit.ListIngre
 		return nil, ErrIngressNotConnected
 	}
 
-	infos, err := s.store.ListIngress(ctx, livekit.RoomName(req.RoomName))
-	if err != nil {
-		logger.Errorw("could not list ingress info", err)
-		return nil, err
+	var infos []*livekit.IngressInfo
+	if req.IngressId != "" {
+		info, err := s.store.LoadIngress(ctx, req.IngressId)
+		if err != nil {
+			return nil, err
+		}
+		infos = []*livekit.IngressInfo{info}
+	} else {
+		infos, err = s.store.ListIngress(ctx, livekit.RoomName(req.RoomName))
+		if err != nil {
+			logger.Errorw("could not list ingress info", err)
+			return nil, err
+		}
 	}
 
 	return &livekit.ListIngressResponse{Items: infos}, nil

pkg/service/signal.go

@@ -112,12 +112,6 @@ type signalService struct {
 }
 
 func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest]) (err error) {
-	// copy the context to prevent a race between the session handler closing
-	// and the delivery of any parting messages from the client. take care to
-	// copy the incoming rpc headers to avoid dropping any session vars.
-	ctx, cancel := context.WithCancel(metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context())))
-	defer cancel()
-
 	req, ok := <-stream.Channel()
 	if !ok {
 		return nil
@@ -139,9 +133,6 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe
 		"connID", ss.ConnectionId,
 	)
 
-	reqChan := routing.NewDefaultMessageChannel(livekit.ConnectionID(ss.ConnectionId))
-	defer reqChan.Close()
-
 	sink := routing.NewSignalMessageSink(routing.SignalSinkParams[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest]{
 		Logger:       l,
 		Stream:       stream,
@@ -149,6 +140,24 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe
 		Writer:       signalResponseMessageWriter{},
 		ConnectionID: livekit.ConnectionID(ss.ConnectionId),
 	})
+	reqChan := routing.NewDefaultMessageChannel(livekit.ConnectionID(ss.ConnectionId))
+
+	go func() {
+		err := routing.CopySignalStreamToMessageChannel[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest](
+			stream,
+			reqChan,
+			signalRequestMessageReader{},
+			r.config,
+		)
+		l.Infow("signal stream closed", "error", err)
+
+		reqChan.Close()
+	}()
+
+	// copy the context to prevent a race between the session handler closing
+	// and the delivery of any parting messages from the client. take care to
+	// copy the incoming rpc headers to avoid dropping any session vars.
+	ctx := metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context()))
 
 	err = r.sessionHandler(ctx, livekit.RoomName(ss.RoomName), *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink)
 	if err != nil {
@@ -156,9 +165,7 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe
 		return
 	}
 
-	err = routing.CopySignalStreamToMessageChannel[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest](stream, reqChan, signalRequestMessageReader{}, r.config)
-	l.Infow("signal stream closed", "error", err)
-
+	stream.Hijack()
 	return
 }
 

pkg/sfu/buffer/rtpstats.go

@@ -750,13 +750,13 @@ func (r *RTPStats) GetRtt() uint32 {
 }
 
 func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) {
-	if srData == nil {
-		return
-	}
-
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
+	if srData == nil || !r.initialized {
+		return
+	}
+
 	// prevent against extreme case of anachronous sender reports
 	if r.srNewest != nil && r.srNewest.NTPTimestamp > srData.NTPTimestamp {
 		r.logger.Infow(
@@ -767,29 +767,6 @@ func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) {
 		return
 	}
 
-	// monitor and log RTP timestamp anomalies
-	var ntpDiffSinceLast time.Duration
-	var rtpDiffSinceLast uint32
-	var arrivalDiffSinceLast time.Duration
-	var expectedTimeDiffSinceLast float64
-	var reason string
-	if r.srNewest != nil {
-		ntpDiffSinceLast = srData.NTPTimestamp.Time().Sub(r.srNewest.NTPTimestamp.Time())
-		rtpDiffSinceLast = srData.RTPTimestamp - r.srNewest.RTPTimestamp
-		arrivalDiffSinceLast = srData.At.Sub(r.srNewest.At)
-
-		expectedTimeDiffSinceLast = float64(rtpDiffSinceLast) / float64(r.params.ClockRate)
-
-		if (srData.RTPTimestamp - r.srNewest.RTPTimestamp) > (1 << 31) {
-			reason = "received sender report, out-of-order" // should not happen, just a sanity check
-		} else {
-			if math.Abs(expectedTimeDiffSinceLast-ntpDiffSinceLast.Seconds()) > 0.2 {
-				// more than 200 ms away from expected delta
-				reason = "received sender report, time warp"
-			}
-		}
-	}
-
 	cycles := uint64(0)
 	if r.srNewest != nil {
 		cycles = r.srNewest.RTPTimestampExt & 0xFF_FF_FF_FF_00_00_00_00
@@ -800,15 +777,49 @@ func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) {
 
 	srDataCopy := *srData
 	srDataCopy.RTPTimestampExt = uint64(srDataCopy.RTPTimestamp) + cycles
+
+	// monitor and log RTP timestamp anomalies
+	var ntpDiffSinceLast time.Duration
+	var rtpDiffSinceLast uint32
+	var arrivalDiffSinceLast time.Duration
+	var expectedTimeDiffSinceLast float64
+	var isWarped bool
+	if r.srNewest != nil {
+		if srDataCopy.RTPTimestampExt < r.srNewest.RTPTimestampExt {
+			// This can happen when a track is replaced with a null and then restored -
+			// i. e. muting replacing with null and unmute restoring the original track.
+			// Under such a condition reset the sender reports to start from this point.
+			// Resetting will ensure sample rate calculations do not go haywire due to negative time.
+			r.logger.Infow(
+				"received sender report, out-of-order, resetting",
+				"prevTSExt", r.srNewest.RTPTimestampExt,
+				"prevNTP", r.srNewest.NTPTimestamp.Time().String(),
+				"currTSExt", srDataCopy.RTPTimestampExt,
+				"currNTP", srDataCopy.NTPTimestamp.Time().String(),
+			)
+			r.srFirst = &srDataCopy
+			r.srNewest = &srDataCopy
+		}
+
+		ntpDiffSinceLast = srDataCopy.NTPTimestamp.Time().Sub(r.srNewest.NTPTimestamp.Time())
+		rtpDiffSinceLast = srDataCopy.RTPTimestamp - r.srNewest.RTPTimestamp
+		arrivalDiffSinceLast = srDataCopy.At.Sub(r.srNewest.At)
+		expectedTimeDiffSinceLast = float64(rtpDiffSinceLast) / float64(r.params.ClockRate)
+		if math.Abs(expectedTimeDiffSinceLast-ntpDiffSinceLast.Seconds()) > 0.2 {
+			// more than 200 ms away from expected delta
+			isWarped = true
+		}
+	}
+
 	r.srNewest = &srDataCopy
 	if r.srFirst == nil {
 		r.srFirst = &srDataCopy
 	}
 
-	if reason != "" {
+	if isWarped {
 		packetDriftResult, reportDriftResult := r.getDrift()
 		r.logger.Infow(
-			reason,
+			"received sender report, time warp",
 			"ntp", srData.NTPTimestamp.Time().String(),
 			"rtp", srData.RTPTimestamp,
 			"arrival", srData.At.String(),
@@ -840,26 +851,22 @@ func (r *RTPStats) GetRtcpSenderReportData() (srFirst *RTCPSenderReportData, srN
 	return
 }
 
-func (r *RTPStats) GetExpectedRTPTimestamp(at time.Time) (uint32, uint64, error) {
+func (r *RTPStats) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt uint64, err error) {
 	r.lock.RLock()
 	defer r.lock.RUnlock()
 
 	if !r.initialized {
-		return 0, 0, errors.New("uninitilaized")
+		err = errors.New("uninitilaized")
+		return
 	}
 
 	timeDiff := at.Sub(r.firstTime)
 	expectedRTPDiff := timeDiff.Nanoseconds() * int64(r.params.ClockRate) / 1e9
-	expectedExtRTP := r.extStartTS + uint64(expectedRTPDiff)
-
-	minTS := ^uint64(0)
-	if r.srNewest != nil {
-		minTS = r.srNewest.RTPTimestampExt
-	}
-	return uint32(expectedExtRTP), minTS, nil
+	expectedTSExt = r.extStartTS + uint64(expectedRTPDiff)
+	return
 }
 
-func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srFirst *RTCPSenderReportData, srNewest *RTCPSenderReportData) *rtcp.SenderReport {
+func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) *rtcp.SenderReport {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
@@ -877,24 +884,42 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srFirst *RTCPSenderReportDat
 
 	// It is possible that publisher is pacing at a slower rate.
 	// That would make `highestTS` to be lagging the RTP time stamp in the RTCP Sender Report from publisher.
-	// Check for that and use the later time stamp if applicable.
+	// Check for that using calculated clock rate and use the later time stamp if applicable.
 	tsCycles := r.tsCycles
 	if nowRTP < r.highestTS {
 		tsCycles++
 	}
 	nowRTPExt := getExtTS(nowRTP, tsCycles)
-	if srFirst != nil && srNewest != nil && srFirst.RTPTimestamp != srNewest.RTPTimestamp {
-		// use incoming rate as a guide
-		tsf := srNewest.NTPTimestamp.Time().Sub(srFirst.NTPTimestamp.Time())
-		rdsf := srNewest.RTPTimestampExt - srFirst.RTPTimestampExt
-		sr := float64(rdsf) / tsf.Seconds()
-		nowRTPExtUsingRate := r.extStartTS + uint64(sr*timeSinceFirst.Seconds())
+	var nowRTPExtUsingRate uint64
+	if calculatedClockRate != 0 {
+		nowRTPExtUsingRate = r.extStartTS + uint64(float64(calculatedClockRate)*timeSinceFirst.Seconds())
 		if nowRTPExtUsingRate > nowRTPExt {
 			nowRTPExt = nowRTPExtUsingRate
 			nowRTP = uint32(nowRTPExtUsingRate)
 		}
 	}
 
+	if r.srNewest != nil && nowRTPExt < r.srNewest.RTPTimestampExt {
+		// If report being generated is behind, use the time different and clock rate of codec to produce next report.
+		// Current report could be behind due to the following
+		//  - Publisher pacing
+		//  - Due to above, report from publisher side is ahead of packet timestamps.
+		//    Note that report will map wall clock to timestamp at capture time and happens before the pacer.
+		//  - Pause/Mute followed by resume, some combination of events that could
+		//    result in this module not having calculated clock rate of publisher side.
+		//  - When the above happens, current will be generated using highestTS which could be behind.
+		//    That could end up behind the last report's timestamp in extreme cases
+		r.logger.Infow(
+			"sending sender report, out-of-order, repairing",
+			"prevTSExt", r.srNewest.RTPTimestampExt,
+			"prevNTP", r.srNewest.NTPTimestamp.Time().String(),
+			"currTSExt", nowRTPExt,
+			"currNTP", nowNTP.Time().String(),
+		)
+		ntpDiffSinceLast := nowNTP.Time().Sub(r.srNewest.NTPTimestamp.Time())
+		nowRTPExt = r.srNewest.RTPTimestampExt + uint64(ntpDiffSinceLast.Seconds()*float64(r.params.ClockRate))
+	}
+
 	// monitor and log RTP timestamp anomalies
 	var ntpDiffSinceLast time.Duration
 	var rtpDiffSinceLast uint32
@@ -938,6 +963,28 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srFirst *RTCPSenderReportDat
 			"reportDrift", reportDriftResult.String(),
 			"highestTS", r.highestTS,
 			"highestTime", r.highestTime.String(),
+			"calculatedClockRate", calculatedClockRate,
+			"nowRTPExt", nowRTPExt,
+			"nowRTPExtUsingRate", nowRTPExtUsingRate,
+		)
+	} else {
+		packetDriftResult, reportDriftResult := r.getDrift()
+		r.logger.Debugw(
+			"sending sender report",
+			"ntp", nowNTP.Time().String(),
+			"rtp", nowRTP,
+			"departure", now.String(),
+			"ntpDiffSinceLast", ntpDiffSinceLast.Seconds(),
+			"rtpDiffSinceLast", int32(rtpDiffSinceLast),
+			"departureDiffSinceLast", departureDiffSinceLast.Seconds(),
+			"expectedTimeDiffSinceLast", expectedTimeDiffSinceLast,
+			"packetDrift", packetDriftResult.String(),
+			"reportDrift", reportDriftResult.String(),
+			"highestTS", r.highestTS,
+			"highestTime", r.highestTime.String(),
+			"calculatedClockRate", calculatedClockRate,
+			"nowRTPExt", nowRTPExt,
+			"nowRTPExtUsingRate", nowRTPExtUsingRate,
 		)
 	}
 

pkg/sfu/downtrack.go

@@ -42,6 +42,8 @@ type TrackSender interface {
 	HandleRTCPSenderReportData(payloadType webrtc.PayloadType, layer int32, srData *buffer.RTCPSenderReportData) error
 }
 
+// -------------------------------------------------------------------
+
 const (
 	RTPPaddingMaxPayloadSize      = 255
 	RTPPaddingEstimatedHeaderSize = 20
@@ -60,6 +62,8 @@ const (
 	maxPaddingOnMuteDuration    = 5 * time.Second
 )
 
+// -------------------------------------------------------------------
+
 var (
 	ErrUnknownKind                       = errors.New("unknown kind of codec")
 	ErrOutOfOrderSequenceNumberCacheMiss = errors.New("out-of-order sequence number not found in cache")
@@ -197,14 +201,13 @@ type DownTrack struct {
 	transceiver               *webrtc.RTPTransceiver
 	writeStream               webrtc.TrackLocalWriter
 	rtcpReader                *buffer.RTCPReader
-	onCloseHandler            func(willBeResumed bool)
-	onBinding                 func(error)
 
 	listenerLock            sync.RWMutex
 	receiverReportListeners []ReceiverReportListener
 
-	bindLock sync.Mutex
-	bound    atomic.Bool
+	bindLock  sync.Mutex
+	bound     atomic.Bool
+	onBinding func(error)
 
 	isClosed             atomic.Bool
 	connected            atomic.Bool
@@ -235,14 +238,13 @@ type DownTrack struct {
 
 	pacer pacer.Pacer
 
-	// update stats
-	onStatsUpdate func(dt *DownTrack, stat *livekit.AnalyticsStat)
+	maxLayerNotifierCh chan struct{}
 
-	// when max subscribed layer changes
+	cbMu                        sync.RWMutex
+	onStatsUpdate               func(dt *DownTrack, stat *livekit.AnalyticsStat)
 	onMaxSubscribedLayerChanged func(dt *DownTrack, layer int32)
-
-	// update rtt
-	onRttUpdate func(dt *DownTrack, rtt uint32)
+	onRttUpdate                 func(dt *DownTrack, rtt uint32)
+	onCloseHandler              func(willBeResumed bool)
 }
 
 // NewDownTrack returns a DownTrack.
@@ -266,17 +268,18 @@ func NewDownTrack(
 	}
 
 	d := &DownTrack{
-		logger:         logger,
-		id:             r.TrackID(),
-		subscriberID:   subID,
-		maxTrack:       mt,
-		streamID:       r.StreamID(),
-		bufferFactory:  bf,
-		receiver:       r,
-		upstreamCodecs: codecs,
-		kind:           kind,
-		codec:          codecs[0].RTPCodecCapability,
-		pacer:          pacer,
+		logger:             logger,
+		id:                 r.TrackID(),
+		subscriberID:       subID,
+		maxTrack:           mt,
+		streamID:           r.StreamID(),
+		bufferFactory:      bf,
+		receiver:           r,
+		upstreamCodecs:     codecs,
+		kind:               kind,
+		codec:              codecs[0].RTPCodecCapability,
+		pacer:              pacer,
+		maxLayerNotifierCh: make(chan struct{}, 20),
 	}
 	d.forwarder = NewForwarder(
 		d.kind,
@@ -307,11 +310,15 @@ func NewDownTrack(
 		Logger:                    d.logger.WithValues("direction", "down"),
 	})
 	d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) {
-		if d.onStatsUpdate != nil {
-			d.onStatsUpdate(d, stat)
+		if onStatsUpdate := d.getOnStatsUpdate(); onStatsUpdate != nil {
+			onStatsUpdate(d, stat)
 		}
 	})
 
+	if d.kind == webrtc.RTPCodecTypeVideo {
+		go d.maxLayerNotifierWorker()
+	}
+
 	return d, nil
 }
 
@@ -541,6 +548,7 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) {
 	if d.IsClosed() || layer == buffer.InvalidLayerSpatial {
 		return
 	}
+
 	interval := 2 * d.rtpStats.GetRtt()
 	if interval < keyFrameIntervalMin {
 		interval = keyFrameIntervalMin
@@ -550,7 +558,13 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) {
 	}
 	ticker := time.NewTicker(time.Duration(interval) * time.Millisecond)
 	defer ticker.Stop()
+
 	for {
+		locked, _ := d.forwarder.CheckSync()
+		if locked {
+			return
+		}
+
 		if d.connected.Load() {
 			d.logger.Debugw("sending PLI for layer lock", "generation", generation, "layer", layer)
 			d.receiver.SendPLI(layer, false)
@@ -565,6 +579,34 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) {
 	}
 }
 
+func (d *DownTrack) postMaxLayerNotifierEvent() {
+	if d.IsClosed() {
+		return
+	}
+
+	select {
+	case d.maxLayerNotifierCh <- struct{}{}:
+	default:
+		d.logger.Warnw("max layer notifier event queue full", nil)
+	}
+}
+
+func (d *DownTrack) maxLayerNotifierWorker() {
+	more := true
+	for more {
+		_, more = <-d.maxLayerNotifierCh
+
+		maxLayerSpatial := buffer.InvalidLayerSpatial
+		if more {
+			maxLayerSpatial = d.forwarder.GetMaxSubscribedSpatial()
+		}
+		if onMaxSubscribedLayerChanged := d.getOnMaxLayerChanged(); onMaxSubscribedLayerChanged != nil {
+			d.logger.Infow("max subscribed layer changed", "maxLayerSpatial", maxLayerSpatial)
+			onMaxSubscribedLayerChanged(d, maxLayerSpatial)
+		}
+	}
+}
+
 // WriteRTP writes an RTP Packet to the DownTrack
 func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 	if !d.bound.Load() || !d.connected.Load() {
@@ -725,17 +767,17 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa
 
 // Mute enables or disables media forwarding - subscriber triggered
 func (d *DownTrack) Mute(muted bool) {
-	changed, maxLayer := d.forwarder.Mute(muted)
-	d.handleMute(muted, false, changed, maxLayer)
+	changed := d.forwarder.Mute(muted)
+	d.handleMute(muted, changed)
 }
 
 // PubMute enables or disables media forwarding - publisher side
 func (d *DownTrack) PubMute(pubMuted bool) {
-	changed, maxLayer := d.forwarder.PubMute(pubMuted)
-	d.handleMute(pubMuted, true, changed, maxLayer)
+	changed := d.forwarder.PubMute(pubMuted)
+	d.handleMute(pubMuted, changed)
 }
 
-func (d *DownTrack) handleMute(muted bool, isPub bool, changed bool, maxLayer buffer.VideoLayer) {
+func (d *DownTrack) handleMute(muted bool, changed bool) {
 	if !changed {
 		return
 	}
@@ -762,18 +804,7 @@ func (d *DownTrack) handleMute(muted bool, isPub bool, changed bool, maxLayer bu
 	// Note that while publisher mute is active, subscriber changes can also happen
 	// and that could turn on/off layers on publisher side.
 	//
-	if !isPub && d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo {
-		notifyLayer := buffer.InvalidLayerSpatial
-		if !muted {
-			//
-			// When unmuting, don't wait for layer lock as
-			// client might need to be notified to start layers
-			// before locking can happen in the forwarder.
-			//
-			notifyLayer = maxLayer.Spatial
-		}
-		d.onMaxSubscribedLayerChanged(d, notifyLayer)
-	}
+	d.postMaxLayerNotifierEvent()
 
 	if sal := d.getStreamAllocatorListener(); sal != nil {
 		sal.OnSubscriptionChanged(d)
@@ -856,12 +887,10 @@ func (d *DownTrack) CloseWithFlush(flush bool) {
 	d.rtpStats.Stop()
 	d.logger.Infow("rtp stats", "direction", "downstream", "mime", d.mime, "ssrc", d.ssrc, "stats", d.rtpStats.ToString())
 
-	if d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo {
-		d.onMaxSubscribedLayerChanged(d, buffer.InvalidLayerSpatial)
-	}
+	close(d.maxLayerNotifierCh)
 
-	if d.onCloseHandler != nil {
-		d.onCloseHandler(!flush)
+	if onCloseHandler := d.getOnCloseHandler(); onCloseHandler != nil {
+		onCloseHandler(!flush)
 	}
 
 	d.stopKeyFrameRequester()
@@ -869,21 +898,12 @@ func (d *DownTrack) CloseWithFlush(flush bool) {
 }
 
 func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) {
-	changed, maxLayer, currentLayer := d.forwarder.SetMaxSpatialLayer(spatialLayer)
+	changed, maxLayer := d.forwarder.SetMaxSpatialLayer(spatialLayer)
 	if !changed {
 		return
 	}
 
-	if d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo && maxLayer.SpatialGreaterThanOrEqual(currentLayer) {
-		//
-		// Notify when new max is
-		//   1. Equal to current -> already locked to the new max
-		//   2. Greater than current -> two scenarios
-		//      a. is higher than previous max -> client may need to start higher layer before forwarder can lock
-		//      b. is lower than previous max -> client can stop higher layer(s)
-		//
-		d.onMaxSubscribedLayerChanged(d, maxLayer.Spatial)
-	}
+	d.postMaxLayerNotifierEvent()
 
 	if sal := d.getStreamAllocatorListener(); sal != nil {
 		sal.OnSubscribedLayerChanged(d, maxLayer)
@@ -891,7 +911,7 @@ func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) {
 }
 
 func (d *DownTrack) SetMaxTemporalLayer(temporalLayer int32) {
-	changed, maxLayer, _ := d.forwarder.SetMaxTemporalLayer(temporalLayer)
+	changed, maxLayer := d.forwarder.SetMaxTemporalLayer(temporalLayer)
 	if !changed {
 		return
 	}
@@ -973,10 +993,23 @@ func (d *DownTrack) UpTrackBitrateReport(availableLayers []int32, bitrates Bitra
 
 // OnCloseHandler method to be called on remote tracked removed
 func (d *DownTrack) OnCloseHandler(fn func(willBeResumed bool)) {
+	d.cbMu.Lock()
+	defer d.cbMu.Unlock()
+
 	d.onCloseHandler = fn
 }
 
+func (d *DownTrack) getOnCloseHandler() func(willBeResumed bool) {
+	d.cbMu.RLock()
+	defer d.cbMu.RUnlock()
+
+	return d.onCloseHandler
+}
+
 func (d *DownTrack) OnBinding(fn func(error)) {
+	d.bindLock.Lock()
+	defer d.bindLock.Unlock()
+
 	d.onBinding = fn
 }
 
@@ -988,17 +1021,47 @@ func (d *DownTrack) AddReceiverReportListener(listener ReceiverReportListener) {
 }
 
 func (d *DownTrack) OnStatsUpdate(fn func(dt *DownTrack, stat *livekit.AnalyticsStat)) {
+	d.cbMu.Lock()
+	defer d.cbMu.Unlock()
+
 	d.onStatsUpdate = fn
 }
 
+func (d *DownTrack) getOnStatsUpdate() func(dt *DownTrack, stat *livekit.AnalyticsStat) {
+	d.cbMu.RLock()
+	defer d.cbMu.RUnlock()
+
+	return d.onStatsUpdate
+}
+
 func (d *DownTrack) OnRttUpdate(fn func(dt *DownTrack, rtt uint32)) {
+	d.cbMu.Lock()
+	defer d.cbMu.Unlock()
+
 	d.onRttUpdate = fn
 }
 
+func (d *DownTrack) getOnRttUpdate() func(dt *DownTrack, rtt uint32) {
+	d.cbMu.RLock()
+	defer d.cbMu.RUnlock()
+
+	return d.onRttUpdate
+}
+
 func (d *DownTrack) OnMaxLayerChanged(fn func(dt *DownTrack, layer int32)) {
+	d.cbMu.Lock()
+	defer d.cbMu.Unlock()
+
 	d.onMaxSubscribedLayerChanged = fn
 }
 
+func (d *DownTrack) getOnMaxLayerChanged() func(dt *DownTrack, layer int32) {
+	d.cbMu.RLock()
+	defer d.cbMu.RUnlock()
+
+	return d.onMaxSubscribedLayerChanged
+}
+
 func (d *DownTrack) IsDeficient() bool {
 	return d.forwarder.IsDeficient()
 }
@@ -1102,8 +1165,11 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport {
 		return nil
 	}
 
-	srFirst, srNewest := d.receiver.GetRTCPSenderReportData(d.forwarder.GetReferenceLayerSpatial())
-	return d.rtpStats.GetRtcpSenderReport(d.ssrc, srFirst, srNewest)
+	clockLayer := d.forwarder.CurrentLayer().Spatial
+	if clockLayer == buffer.InvalidLayerSpatial {
+		clockLayer = d.forwarder.GetReferenceLayerSpatial()
+	}
+	return d.rtpStats.GetRtcpSenderReport(d.ssrc, d.receiver.GetCalculatedClockRate(clockLayer))
 }
 
 func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan struct{} {
@@ -1356,8 +1422,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 			d.sequencer.setRTT(rttToReport)
 		}
 
-		if d.onRttUpdate != nil {
-			d.onRttUpdate(d, rttToReport)
+		if onRttUpdate := d.getOnRttUpdate(); onRttUpdate != nil {
+			onRttUpdate(d, rttToReport)
 		}
 	}
 }
@@ -1542,7 +1608,7 @@ func (d *DownTrack) DebugInfo() map[string]interface{} {
 	}
 }
 
-func (d *DownTrack) getExpectedRTPTimestamp(at time.Time) (uint32, uint64, error) {
+func (d *DownTrack) getExpectedRTPTimestamp(at time.Time) (uint64, error) {
 	return d.rtpStats.GetExpectedRTPTimestamp(at)
 }
 
@@ -1745,15 +1811,8 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int,
 	}
 
 	if spmd.tp != nil {
-		if spmd.tp.isSwitchingToMaxSpatial && d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo {
-			d.onMaxSubscribedLayerChanged(d, spmd.tp.maxSpatialLayer)
-		}
-
-		if spmd.tp.isSwitchingToRequestSpatial {
-			locked, _ := d.forwarder.CheckSync()
-			if locked {
-				d.stopKeyFrameRequester()
-			}
+		if spmd.tp.isSwitching {
+			d.postMaxLayerNotifierEvent()
 		}
 
 		if spmd.tp.isResuming {

pkg/sfu/forwarder.go

@@ -26,6 +26,10 @@ const (
 	FlagFilterRTX           = true
 	TransitionCostSpatial   = 10
 	ParkedLayerWaitDuration = 2 * time.Second
+
+	ResumeBehindThresholdSeconds      = float64(0.1)   // 100ms
+	LayerSwitchBehindThresholdSeconds = float64(0.05)  // 50ms
+	SwitchAheadThresholdSeconds       = float64(0.025) // 25ms
 )
 
 // -------------------------------------------------------------------
@@ -125,15 +129,13 @@ func (v VideoTransition) String() string {
 // -------------------------------------------------------------------
 
 type TranslationParams struct {
-	shouldDrop                  bool
-	isResuming                  bool
-	isSwitchingToRequestSpatial bool
-	isSwitchingToMaxSpatial     bool
-	maxSpatialLayer             int32
-	rtp                         *TranslationParamsRTP
-	codecBytes                  []byte
-	ddBytes                     []byte
-	marker                      bool
+	shouldDrop  bool
+	isResuming  bool
+	isSwitching bool
+	rtp         *TranslationParamsRTP
+	codecBytes  []byte
+	ddBytes     []byte
+	marker      bool
 }
 
 // -------------------------------------------------------------------
@@ -171,7 +173,7 @@ type Forwarder struct {
 	kind                          webrtc.RTPCodecType
 	logger                        logger.Logger
 	getReferenceLayerRTPTimestamp func(ts uint32, layer int32, referenceLayer int32) (uint32, error)
-	getExpectedRTPTimestamp       func(at time.Time) (uint32, uint64, error)
+	getExpectedRTPTimestamp       func(at time.Time) (uint64, error)
 
 	muted    bool
 	pubMuted bool
@@ -202,7 +204,7 @@ func NewForwarder(
 	kind webrtc.RTPCodecType,
 	logger logger.Logger,
 	getReferenceLayerRTPTimestamp func(ts uint32, layer int32, referenceLayer int32) (uint32, error),
-	getExpectedRTPTimestamp func(at time.Time) (uint32, uint64, error),
+	getExpectedRTPTimestamp func(at time.Time) (uint64, error),
 ) *Forwarder {
 	f := &Forwarder{
 		kind:                          kind,
@@ -356,12 +358,12 @@ func (f *Forwarder) SeedState(state ForwarderState) {
 	f.refTSOffset = state.RefTSOffset
 }
 
-func (f *Forwarder) Mute(muted bool) (bool, buffer.VideoLayer) {
+func (f *Forwarder) Mute(muted bool) bool {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
 	if f.muted == muted {
-		return false, f.vls.GetMax()
+		return false
 	}
 
 	// Do not mute when paused due to bandwidth limitation.
@@ -380,7 +382,7 @@ func (f *Forwarder) Mute(muted bool) (bool, buffer.VideoLayer) {
 	// the case of intentional mute.
 	if muted && f.isDeficientLocked() && f.lastAllocation.PauseReason == VideoPauseReasonBandwidth {
 		f.logger.Infow("ignoring forwarder mute, paused due to congestion")
-		return false, f.vls.GetMax()
+		return false
 	}
 
 	f.logger.Debugw("setting forwarder mute", "muted", muted)
@@ -391,7 +393,7 @@ func (f *Forwarder) Mute(muted bool) (bool, buffer.VideoLayer) {
 		f.resyncLocked()
 	}
 
-	return true, f.vls.GetMax()
+	return true
 }
 
 func (f *Forwarder) IsMuted() bool {
@@ -401,12 +403,12 @@ func (f *Forwarder) IsMuted() bool {
 	return f.muted
 }
 
-func (f *Forwarder) PubMute(pubMuted bool) (bool, buffer.VideoLayer) {
+func (f *Forwarder) PubMute(pubMuted bool) bool {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
 	if f.pubMuted == pubMuted {
-		return false, f.vls.GetMax()
+		return false
 	}
 
 	f.logger.Debugw("setting forwarder pub mute", "pubMuted", pubMuted)
@@ -428,7 +430,7 @@ func (f *Forwarder) PubMute(pubMuted bool) (bool, buffer.VideoLayer) {
 		}
 	}
 
-	return true, f.vls.GetMax()
+	return true
 }
 
 func (f *Forwarder) IsPubMuted() bool {
@@ -445,17 +447,17 @@ func (f *Forwarder) IsAnyMuted() bool {
 	return f.muted || f.pubMuted
 }
 
-func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLayer, buffer.VideoLayer) {
+func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLayer) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
 	if f.kind == webrtc.RTPCodecTypeAudio {
-		return false, buffer.InvalidLayer, buffer.InvalidLayer
+		return false, buffer.InvalidLayer
 	}
 
 	existingMax := f.vls.GetMax()
 	if spatialLayer == existingMax.Spatial {
-		return false, existingMax, f.vls.GetCurrent()
+		return false, existingMax
 	}
 
 	f.logger.Debugw("setting max spatial layer", "layer", spatialLayer)
@@ -463,20 +465,20 @@ func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLa
 
 	f.clearParkedLayer()
 
-	return true, f.vls.GetMax(), f.vls.GetCurrent()
+	return true, f.vls.GetMax()
 }
 
-func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.VideoLayer, buffer.VideoLayer) {
+func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.VideoLayer) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
 	if f.kind == webrtc.RTPCodecTypeAudio {
-		return false, buffer.InvalidLayer, buffer.InvalidLayer
+		return false, buffer.InvalidLayer
 	}
 
 	existingMax := f.vls.GetMax()
 	if temporalLayer == existingMax.Temporal {
-		return false, existingMax, f.vls.GetCurrent()
+		return false, existingMax
 	}
 
 	f.logger.Debugw("setting max temporal layer", "layer", temporalLayer)
@@ -484,7 +486,7 @@ func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.Video
 
 	f.clearParkedLayer()
 
-	return true, f.vls.GetMax(), f.vls.GetCurrent()
+	return true, f.vls.GetMax()
 }
 
 func (f *Forwarder) MaxLayer() buffer.VideoLayer {
@@ -508,6 +510,25 @@ func (f *Forwarder) TargetLayer() buffer.VideoLayer {
 	return f.vls.GetTarget()
 }
 
+func (f *Forwarder) GetMaxSubscribedSpatial() int32 {
+	f.lock.RLock()
+	defer f.lock.RUnlock()
+
+	layer := buffer.InvalidLayerSpatial // covers muted case
+	if !f.muted {
+		layer = f.vls.GetMax().Spatial
+
+		// If current is higher, mark the current layer as max subscribed layer
+		// to prevent the current layer from stopping before forwarder switches
+		// to the new and lower max layer,
+		if layer < f.vls.GetCurrent().Spatial {
+			layer = f.vls.GetCurrent().Spatial
+		}
+	}
+
+	return layer
+}
+
 func (f *Forwarder) GetReferenceLayerSpatial() int32 {
 	f.lock.RLock()
 	defer f.lock.RUnlock()
@@ -1477,103 +1498,176 @@ func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32)
 	return nil, ErrUnknownKind
 }
 
+func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) error {
+	if !f.started {
+		f.started = true
+		f.referenceLayerSpatial = layer
+		f.rtpMunger.SetLastSnTs(extPkt)
+		f.codecMunger.SetLast(extPkt)
+		f.logger.Infow(
+			"starting forwarding",
+			"sequenceNumber", extPkt.Packet.SequenceNumber,
+			"timestamp", extPkt.Packet.Timestamp,
+			"layer", layer,
+			"referenceLayerSpatial", f.referenceLayerSpatial,
+		)
+		return nil
+	}
+
+	if f.referenceLayerSpatial == buffer.InvalidLayerSpatial {
+		// on a resume, reference layer may not be set, so only set when it is invalid
+		f.referenceLayerSpatial = layer
+	}
+
+	// Compute how much time passed between the previous forwarded packet
+	// and the current incoming (to be forwarded) packet and calculate
+	// timestamp offset on source change.
+	//
+	// There are three timestamps to consider here
+	//   1. lastTS -> timestamp of last sent packet
+	//   2. refTS -> timestamp of this packet (after munging) calculated using feed's RTCP sender report
+	//   3. expectedTS -> expected timestamp of this packet calculated based on elapsed time since first packet
+	// Ideally, refTS and expectedTS should be very close and lastTS should be before both of those.
+	// But, cases like muting/unmuting, clock vagaries, pacing, etc. make them not satisfy those conditions always.
+	lastTS := f.rtpMunger.GetLast().LastTS
+	refTS := lastTS
+	expectedTS := lastTS
+	switchingAt := time.Now()
+	if f.getReferenceLayerRTPTimestamp != nil {
+		ts, err := f.getReferenceLayerRTPTimestamp(extPkt.Packet.Timestamp, layer, f.referenceLayerSpatial)
+		if err == nil {
+			refTS = ts
+		}
+		// AVSYNC-TODO: can error out here if refTS is not available. It can happen when there is no sender report
+		// for the layer being switched to. Can especially happen at the start of the track when layer switches are
+		// potentially happening very quickly. Erroring out and waiting for a layer for which a sender report has been
+		// received will calculate a better offset, but may result in initial adaptation to take a bit longer depending
+		// on how often publisher/remote side sends RTCP sender report.
+	}
+
+	if f.getExpectedRTPTimestamp != nil {
+		tsExt, err := f.getExpectedRTPTimestamp(switchingAt)
+		if err == nil {
+			expectedTS = uint32(tsExt)
+		} else {
+			rtpDiff := uint32(0)
+			if !f.preStartTime.IsZero() && f.refTSOffset == 0 {
+				timeSinceFirst := time.Since(f.preStartTime)
+				rtpDiff = uint32(timeSinceFirst.Nanoseconds() * int64(f.codec.ClockRate) / 1e9)
+				f.refTSOffset = f.firstTS + rtpDiff - refTS
+				f.logger.Infow(
+					"calculating refTSOffset",
+					"preStartTime", f.preStartTime.String(),
+					"firstTS", f.firstTS,
+					"timeSinceFirst", timeSinceFirst,
+					"rtpDiff", rtpDiff,
+					"refTS", refTS,
+					"refTSOffset", f.refTSOffset,
+				)
+			}
+			expectedTS += rtpDiff
+		}
+	}
+	refTS += f.refTSOffset
+
+	var nextTS uint32
+	if f.lastSSRC == 0 {
+		// If resuming (e. g. on unmute), keep next timestamp close to expected timestamp.
+		//
+		// Rationale:
+		// Case 1: If mute is implemented via something like stopping a track and resuming it on unmute,
+		// the RTP timestamp may not have jumped across mute valley. In this case, old timestamp
+		// should not be used.
+		//
+		// Case 2: OTOH, something like pacing may be adding latency in the publisher path (even if
+		// the timestamps incremented correctly across the mute valley). In this case, reference
+		// timestamp should be used as things will catch up to real time when channel capacity
+		// increases and pacer starts sending at faster rate.
+		//
+		// But, the challenege is distinguishing between the two cases. As a compromise, the difference
+		// between expectedTS and refTS is thresholded. Difference below the threshold is treated as Case 2
+		// and above as Case 1.
+		//
+		// In the event of refTS > expectedTS, another threshold is used to pick the next timestamp.
+		// Ideally, refTS should not be ahead of expectedTS, but expectedTS uses the first packet's
+		// wall clock time. So, if the first packet experienced abmormal latency, it is possible
+		// for refTS > expectedTS
+		diffSeconds := float64(expectedTS-refTS) / float64(f.codec.ClockRate)
+		if diffSeconds >= 0.0 {
+			if diffSeconds > ResumeBehindThresholdSeconds {
+				f.logger.Infow("resume, reference too far behind", "expectedTS", expectedTS, "refTS", refTS, "diffSeconds", diffSeconds)
+				nextTS = expectedTS
+			} else {
+				nextTS = refTS
+			}
+		} else {
+			if math.Abs(diffSeconds) > SwitchAheadThresholdSeconds {
+				f.logger.Infow("resume, reference too far ahead", "expectedTS", expectedTS, "refTS", refTS, "diffSeconds", math.Abs(diffSeconds))
+				nextTS = expectedTS
+			} else {
+				nextTS = refTS
+			}
+		}
+	} else {
+		// switching between layers, check if refTS is too far behind the last sent
+		diffSeconds := float64(refTS-lastTS) / float64(f.codec.ClockRate)
+		if diffSeconds < 0.0 {
+			if math.Abs(diffSeconds) > LayerSwitchBehindThresholdSeconds {
+				// AVSYNC-TODO: This could be due to pacer trickling out this layer. Should potentially return error here and wait for a more opportune time
+				// or some forcing function (like "have waited for too long for layer switch, nothing available, switch to whatever is available" kind of condition)
+				// to do the switch. Just logging it for now.
+				f.logger.Infow("layer switch, reference too far behind", "expectedTS", expectedTS, "refTS", refTS, "lastTS", lastTS, "diffSeconds", math.Abs(diffSeconds))
+			}
+			// use a nominal increase to ensure that timestamp is always moving forward
+			nextTS = lastTS + 1
+		} else {
+			diffSeconds = float64(expectedTS-refTS) / float64(f.codec.ClockRate)
+			if diffSeconds < 0.0 && math.Abs(diffSeconds) > SwitchAheadThresholdSeconds {
+				f.logger.Infow("layer switch, reference too far ahead", "expectedTS", expectedTS, "refTS", refTS, "diffSeconds", math.Abs(diffSeconds))
+				nextTS = expectedTS
+			} else {
+				nextTS = refTS
+			}
+		}
+	}
+
+	if nextTS-lastTS == 0 || nextTS-lastTS > (1<<31) {
+		f.logger.Infow("next timestamp is before last, adjusting", "nextTS", nextTS, "lastTS", lastTS)
+		// nominal increase
+		nextTS = lastTS + 1
+	}
+	f.logger.Infow(
+		"next timestamp on switch",
+		"switchingAt", switchingAt.String(),
+		"layer", layer,
+		"lastTS", lastTS,
+		"refTS", refTS,
+		"refTSOffset", f.refTSOffset,
+		"referenceLayerSpatial", f.referenceLayerSpatial,
+		"expectedTS", expectedTS,
+		"nextTS", nextTS,
+		"jump", nextTS-lastTS,
+	)
+
+	f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, nextTS-lastTS)
+	f.codecMunger.UpdateOffsets(extPkt)
+	return nil
+}
+
 // should be called with lock held
 func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, layer int32, tp *TranslationParams) (*TranslationParams, error) {
+	if tp == nil {
+		tp = &TranslationParams{}
+	}
 	if f.lastSSRC != extPkt.Packet.SSRC {
-		if !f.started {
-			f.started = true
-			f.referenceLayerSpatial = layer
-			f.rtpMunger.SetLastSnTs(extPkt)
-			f.codecMunger.SetLast(extPkt)
-			f.logger.Infow(
-				"starting forwarding",
-				"sequenceNumber", extPkt.Packet.SequenceNumber,
-				"timestamp", extPkt.Packet.Timestamp,
-				"layer", layer,
-				"referenceLayerSpatial", f.referenceLayerSpatial,
-			)
-		} else {
-			if f.referenceLayerSpatial == buffer.InvalidLayerSpatial {
-				// on a resume, reference layer may not be set, so only set when it is invalid
-				f.referenceLayerSpatial = layer
-			}
-
-			// Compute how much time passed between the old RTP extPkt
-			// and the current packet, and fix timestamp on source change
-			//
-			// There are three time stamps to consider here
-			//   1. lastTS -> time stamp of last sent packet
-			//   2. refTS -> time stamp of this packet (after munging) calculated using feed's RTCP sender report
-			//   3. expectedTS -> time stamp of this packet (after munging) calculated using this stream's RTCP sender report
-			// Ideally, refTS and expectedTS should be very close and lastTS should be before both of those.
-			// But, cases like muting/unmuting, clock vagaries make them not satisfy those conditions always.
-			//
-			// There are 6 orderings to consider (considering only inequalities). Resolve them using following rules
-			//   1. Timestamp has to move forward
-			//   2. Keep next time stamp close to expected
-			lastTS := f.rtpMunger.GetLast().LastTS
-			refTS := lastTS
-			expectedTS := lastTS
-			minTS := ^uint64(0)
-			switchingAt := time.Now()
-			if f.getReferenceLayerRTPTimestamp != nil {
-				ts, err := f.getReferenceLayerRTPTimestamp(extPkt.Packet.Timestamp, layer, f.referenceLayerSpatial)
-				if err == nil {
-					refTS = ts
-				}
-			}
-			if f.getExpectedRTPTimestamp != nil {
-				ts, min, err := f.getExpectedRTPTimestamp(switchingAt)
-				if err == nil {
-					expectedTS = ts
-					minTS = min
-				} else {
-					rtpDiff := uint32(0)
-					if !f.preStartTime.IsZero() && f.refTSOffset == 0 {
-						timeSinceFirst := time.Since(f.preStartTime)
-						rtpDiff = uint32(timeSinceFirst.Nanoseconds() * int64(f.codec.ClockRate) / 1e9)
-						f.refTSOffset = f.firstTS + rtpDiff - refTS
-						f.logger.Infow(
-							"calculating refTSOffset",
-							"preStartTime", f.preStartTime.String(),
-							"firstTS", f.firstTS,
-							"timeSinceFirst", timeSinceFirst,
-							"rtpDiff", rtpDiff,
-							"refTS", refTS,
-							"refTSOffset", f.refTSOffset,
-						)
-					}
-					expectedTS += rtpDiff
-				}
-			}
-			refTS += f.refTSOffset
-			nextTS, explain := getNextTimestamp(lastTS, refTS, expectedTS, minTS)
-			f.logger.Infow(
-				"next timestamp on switch",
-				"switchingAt", switchingAt.String(),
-				"layer", layer,
-				"lastTS", lastTS,
-				"refTS", refTS,
-				"refTSOffset", f.refTSOffset,
-				"referenceLayerSpatial", f.referenceLayerSpatial,
-				"expectedTS", expectedTS,
-				"minTS", minTS,
-				"nextTS", nextTS,
-				"jump", nextTS-lastTS,
-				"explanation", explain,
-			)
-
-			f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, nextTS-lastTS)
-			f.codecMunger.UpdateOffsets(extPkt)
+		if err := f.processSourceSwitch(extPkt, layer); err != nil {
+			tp.shouldDrop = true
+			return tp, err
 		}
-
 		f.logger.Debugw("switching feed", "from", f.lastSSRC, "to", extPkt.Packet.SSRC)
 		f.lastSSRC = extPkt.Packet.SSRC
 	}
 
-	if tp == nil {
-		tp = &TranslationParams{}
-	}
 	tpRTP, err := f.rtpMunger.UpdateAndGetSnTs(extPkt)
 	if err != nil {
 		tp.shouldDrop = true
@@ -1620,9 +1714,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in
 		return tp, nil
 	}
 	tp.isResuming = result.IsResuming
-	tp.isSwitchingToRequestSpatial = result.IsSwitchingToRequestSpatial
-	tp.isSwitchingToMaxSpatial = result.IsSwitchingToMaxSpatial
-	tp.maxSpatialLayer = result.MaxSpatialLayer
+	tp.isSwitching = result.IsSwitching
 	tp.ddBytes = result.DependencyDescriptorExtension
 	tp.marker = result.RTPMarker
 
@@ -1696,7 +1788,7 @@ func (f *Forwarder) maybeStart() {
 		Packet: &rtp.Packet{
 			Header: rtp.Header{
 				SequenceNumber: uint16(rand.Intn(1<<14)) + uint16(1<<15), // a random number in third quartile of sequence number space
-				Timestamp:      uint32(rand.Intn(1<<30)) + uint32(1<<31), // a random number in third quartile of time stamp space
+				Timestamp:      uint32(rand.Intn(1<<30)) + uint32(1<<31), // a random number in third quartile of timestamp space
 			},
 		},
 	}
@@ -1741,16 +1833,16 @@ func (f *Forwarder) GetSnTsForBlankFrames(frameRate uint32, numPackets int) ([]S
 
 	lastTS := f.rtpMunger.GetLast().LastTS
 	expectedTS := lastTS
-	minTS := ^uint64(0)
 	if f.getExpectedRTPTimestamp != nil {
-		ts, min, err := f.getExpectedRTPTimestamp(time.Now())
+		tsExt, err := f.getExpectedRTPTimestamp(time.Now())
 		if err == nil {
-			expectedTS = ts
-			minTS = min
+			expectedTS = uint32(tsExt)
 		}
 	}
-	nextTS, _ := getNextTimestamp(lastTS, expectedTS, expectedTS, minTS)
-	snts, err := f.rtpMunger.UpdateAndGetPaddingSnTs(numPackets, f.codec.ClockRate, frameRate, frameEndNeeded, nextTS)
+	if expectedTS-lastTS == 0 || expectedTS-lastTS > (1<<31) {
+		expectedTS = lastTS + 1
+	}
+	snts, err := f.rtpMunger.UpdateAndGetPaddingSnTs(numPackets, f.codec.ClockRate, frameRate, frameEndNeeded, expectedTS)
 	return snts, frameEndNeeded, err
 }
 
@@ -1885,47 +1977,20 @@ done:
 	if !targetLayer.IsValid() {
 		distance += (maxSeenLayer.Temporal + 1)
 	}
+	// TODO-REMOVE-AFTER-DEBUG
+	logger.Debugw(
+		"distance to desired",
+		"maxSeenLauer", maxSeenLayer,
+		"availableLayers", availableLayers,
+		"brs", brs,
+		"targetLayer", targetLayer,
+		"maxLayer", maxLayer,
+		"adjustedMaxLayer", adjustedMaxLayer,
+		"maxAvailableSpatial", maxAvailableSpatial,
+		"maxAvailableTemporal", maxAvailableTemporal,
+		"distance", distance,
+		"distanceToDesired", float64(distance)/float64(maxSeenLayer.Temporal+1),
+	)
 
 	return float64(distance) / float64(maxSeenLayer.Temporal+1)
 }
-
-func getNextTimestamp(lastTS uint32, refTS uint32, expectedTS uint32, minTS uint64) (uint32, string) {
-	isInOrder := func(val1, val2 uint32) bool {
-		diff := val1 - val2
-		return diff != 0 && diff < (1<<31)
-	}
-
-	rl := isInOrder(refTS, lastTS)
-	el := isInOrder(expectedTS, lastTS)
-	er := isInOrder(expectedTS, refTS)
-
-	nextTS := lastTS + 1
-	explain := "l = r = e"
-
-	switch {
-	case rl && el && er: // lastTS < refTS < expectedTS
-		nextTS = uint32(float64(refTS) + 0.05*float64(expectedTS-refTS))
-		explain = fmt.Sprintf("l < r < e, %d, %d", refTS-lastTS, expectedTS-refTS)
-	case rl && el && !er: // lastTS < expectedTS < refTS
-		nextTS = uint32(float64(expectedTS) + 0.5*float64(refTS-expectedTS))
-		explain = fmt.Sprintf("l < e < r, %d, %d", expectedTS-lastTS, refTS-expectedTS)
-	case !rl && el && er: // refTS < lastTS < expectedTS
-		nextTS = uint32(float64(lastTS) + 0.5*float64(expectedTS-lastTS))
-		explain = fmt.Sprintf("r < l < e, %d, %d", lastTS-refTS, expectedTS-lastTS)
-	case !rl && !el && er: // refTS < expectedTS < lastTS
-		nextTS = lastTS + 1
-		explain = fmt.Sprintf("r < e < l, %d, %d", expectedTS-refTS, lastTS-expectedTS)
-	case rl && !el && !er: // expectedTS < lastTS < refTS
-		nextTS = uint32(float64(lastTS) + 0.75*float64(refTS-lastTS))
-		explain = fmt.Sprintf("e < l < r, %d, %d", lastTS-expectedTS, refTS-lastTS)
-	case !rl && !el && !er: // expectedTS < refTS < lastTS
-		nextTS = lastTS + 1
-		explain = fmt.Sprintf("e < r < l, %d, %d", refTS-expectedTS, lastTS-refTS)
-	}
-
-	if minTS != ^uint64(0) && !isInOrder(nextTS, uint32(minTS)) {
-		nextTS = uint32(minTS) + 1
-	}
-
-	return nextTS, explain
-}

pkg/sfu/forwarder_test.go

@@ -26,13 +26,13 @@ func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Fo
 func TestForwarderMute(t *testing.T) {
 	f := newForwarder(testutils.TestOpusCodec, webrtc.RTPCodecTypeAudio)
 	require.False(t, f.IsMuted())
-	muted, _ := f.Mute(false)
+	muted := f.Mute(false)
 	require.False(t, muted) // no change in mute state
 	require.False(t, f.IsMuted())
-	muted, _ = f.Mute(true)
+	muted = f.Mute(true)
 	require.True(t, muted)
 	require.True(t, f.IsMuted())
-	muted, _ = f.Mute(false)
+	muted = f.Mute(false)
 	require.True(t, muted)
 	require.False(t, f.IsMuted())
 }
@@ -45,15 +45,13 @@ func TestForwarderLayersAudio(t *testing.T) {
 	require.Equal(t, buffer.InvalidLayer, f.CurrentLayer())
 	require.Equal(t, buffer.InvalidLayer, f.TargetLayer())
 
-	changed, maxLayer, currentLayer := f.SetMaxSpatialLayer(1)
+	changed, maxLayer := f.SetMaxSpatialLayer(1)
 	require.False(t, changed)
 	require.Equal(t, buffer.InvalidLayer, maxLayer)
-	require.Equal(t, buffer.InvalidLayer, currentLayer)
 
-	changed, maxLayer, currentLayer = f.SetMaxTemporalLayer(1)
+	changed, maxLayer = f.SetMaxTemporalLayer(1)
 	require.False(t, changed)
 	require.Equal(t, buffer.InvalidLayer, maxLayer)
-	require.Equal(t, buffer.InvalidLayer, currentLayer)
 
 	require.Equal(t, buffer.InvalidLayer, f.MaxLayer())
 }
@@ -72,12 +70,11 @@ func TestForwarderLayersVideo(t *testing.T) {
 		Spatial:  buffer.DefaultMaxLayerSpatial,
 		Temporal: buffer.DefaultMaxLayerTemporal,
 	}
-	changed, maxLayer, currentLayer := f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial)
+	changed, maxLayer := f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial)
 	require.True(t, changed)
 	require.Equal(t, expectedLayers, maxLayer)
-	require.Equal(t, buffer.InvalidLayer, currentLayer)
 
-	changed, maxLayer, currentLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1)
+	changed, maxLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1)
 	require.True(t, changed)
 	expectedLayers = buffer.VideoLayer{
 		Spatial:  buffer.DefaultMaxLayerSpatial - 1,
@@ -85,21 +82,18 @@ func TestForwarderLayersVideo(t *testing.T) {
 	}
 	require.Equal(t, expectedLayers, maxLayer)
 	require.Equal(t, expectedLayers, f.MaxLayer())
-	require.Equal(t, buffer.InvalidLayer, currentLayer)
 
 	f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 1})
-	changed, maxLayer, currentLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1)
+	changed, maxLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1)
 	require.False(t, changed)
 	require.Equal(t, expectedLayers, maxLayer)
 	require.Equal(t, expectedLayers, f.MaxLayer())
-	require.Equal(t, buffer.VideoLayer{Spatial: 0, Temporal: 1}, currentLayer)
 
-	changed, maxLayer, currentLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal)
+	changed, maxLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal)
 	require.False(t, changed)
 	require.Equal(t, expectedLayers, maxLayer)
-	require.Equal(t, buffer.VideoLayer{Spatial: 0, Temporal: 1}, currentLayer)
 
-	changed, maxLayer, currentLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal - 1)
+	changed, maxLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal - 1)
 	require.True(t, changed)
 	expectedLayers = buffer.VideoLayer{
 		Spatial:  buffer.DefaultMaxLayerSpatial - 1,
@@ -107,7 +101,6 @@ func TestForwarderLayersVideo(t *testing.T) {
 	}
 	require.Equal(t, expectedLayers, maxLayer)
 	require.Equal(t, expectedLayers, f.MaxLayer())
-	require.Equal(t, buffer.VideoLayer{Spatial: 0, Temporal: 1}, currentLayer)
 }
 
 func TestForwarderAllocateOptimal(t *testing.T) {
@@ -1404,8 +1397,8 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) {
 	marshalledVP8, err := expectedVP8.Marshal()
 	require.NoError(t, err)
 	expectedTP = TranslationParams{
-		isSwitchingToMaxSpatial: true,
-		isResuming:              true,
+		isSwitching: true,
+		isResuming:  true,
 		rtp: &TranslationParamsRTP{
 			snOrdering:     SequenceNumberOrderingContiguous,
 			sequenceNumber: 23333,
@@ -1716,8 +1709,7 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) {
 	marshalledVP8, err = expectedVP8.Marshal()
 	require.NoError(t, err)
 	expectedTP = TranslationParams{
-		isSwitchingToMaxSpatial: true,
-		maxSpatialLayer:         1,
+		isSwitching: true,
 		rtp: &TranslationParamsRTP{
 			snOrdering:     SequenceNumberOrderingContiguous,
 			sequenceNumber: 23339,

pkg/sfu/receiver.go

@@ -66,7 +66,7 @@ type TrackReceiver interface {
 
 	GetTemporalLayerFpsForSpatial(layer int32) []float32
 
-	GetRTCPSenderReportData(layer int32) (*buffer.RTCPSenderReportData, *buffer.RTCPSenderReportData)
+	GetCalculatedClockRate(layer int32) uint32
 	GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error)
 }
 
@@ -752,8 +752,8 @@ func (w *WebRTCReceiver) GetTemporalLayerFpsForSpatial(layer int32) []float32 {
 	return b.GetTemporalLayerFpsForSpatial(layer)
 }
 
-func (w *WebRTCReceiver) GetRTCPSenderReportData(layer int32) (*buffer.RTCPSenderReportData, *buffer.RTCPSenderReportData) {
-	return w.streamTrackerManager.GetRTCPSenderReportData(layer)
+func (w *WebRTCReceiver) GetCalculatedClockRate(layer int32) uint32 {
+	return w.streamTrackerManager.GetCalculatedClockRate(layer)
 }
 
 func (w *WebRTCReceiver) GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) {

pkg/sfu/streamallocator/streamallocator.go

@@ -550,6 +550,10 @@ func (s *StreamAllocator) postEvent(event Event) {
 
 func (s *StreamAllocator) processEvents() {
 	for event := range s.eventCh {
+		if s.isStopped.Load() {
+			break
+		}
+
 		s.handleEvent(&event)
 	}
 

pkg/sfu/streamtrackermanager.go

@@ -2,6 +2,7 @@ package sfu
 
 import (
 	"fmt"
+	"math"
 	"sort"
 	"sync"
 	"time"
@@ -15,6 +16,14 @@ import (
 	"github.com/livekit/protocol/logger"
 )
 
+const (
+	senderReportThresholdSeconds = float64(60.0)
+
+	minDurationForClockRateCalculation = 15 * time.Second
+)
+
+// ---------------------------------------------------
+
 type StreamTrackerManagerListener interface {
 	OnAvailableLayersChanged()
 	OnBitrateAvailabilityChanged()
@@ -24,9 +33,12 @@ type StreamTrackerManagerListener interface {
 	OnBitrateReport(availableLayers []int32, bitrates Bitrates)
 }
 
+// ---------------------------------------------------
+
 type endsSenderReport struct {
-	first  *buffer.RTCPSenderReportData
-	newest *buffer.RTCPSenderReportData
+	first       *buffer.RTCPSenderReportData
+	newest      *buffer.RTCPSenderReportData
+	lastUpdated time.Time
 }
 
 type StreamTrackerManager struct {
@@ -50,6 +62,7 @@ type StreamTrackerManager struct {
 
 	senderReportMu sync.RWMutex
 	senderReports  [buffer.DefaultMaxLayerSpatial + 1]endsSenderReport
+	layerOffsets   [buffer.DefaultMaxLayerSpatial + 1][buffer.DefaultMaxLayerSpatial + 1]uint32
 
 	closed core.Fuse
 
@@ -517,6 +530,40 @@ func (s *StreamTrackerManager) maxExpectedLayerFromTrackInfo() {
 	}
 }
 
+func (s *StreamTrackerManager) updateLayerOffsetLocked(ref, other int32) {
+	srRef := s.senderReports[ref].newest
+	srOther := s.senderReports[other].newest
+	if srRef == nil || srRef.NTPTimestamp == 0 || srOther == nil || srOther.NTPTimestamp == 0 {
+		return
+	}
+
+	ntpDiff := srRef.NTPTimestamp.Time().Sub(srOther.NTPTimestamp.Time())
+	if math.Abs(ntpDiff.Seconds()) > senderReportThresholdSeconds {
+		// offset is updated only if the layers' sender reports are close enough.
+		//
+		// Rationale: higher layers could be paused for extended periods of time
+		// due to adaptive stream/dynacast or publisher constraints like CPU/bandwidth.
+		// The check is to avoid using very old reports.
+		return
+	}
+	rtpDiff := ntpDiff.Nanoseconds() * int64(s.clockRate) / 1e9
+
+	// calculate other layer's time stamp at the same time as ref layer's NTP time
+	normalizedOtherTS := srOther.RTPTimestamp + uint32(rtpDiff)
+
+	// now both layers' time stamp refer to the same NTP time and the diff is the offset between the layers
+	offset := srRef.RTPTimestamp - normalizedOtherTS
+
+	// use minimal offset to indicate value availability in the extremely unlikely case of
+	// both layers using the same timestamp
+	if offset == 0 {
+		s.logger.Infow("using default offset", "ref", ref, "other", other)
+		offset = 1
+	}
+
+	s.layerOffsets[ref][other] = offset
+}
+
 func (s *StreamTrackerManager) SetRTCPSenderReportData(layer int32, srFirst *buffer.RTCPSenderReportData, srNewest *buffer.RTCPSenderReportData) {
 	s.senderReportMu.Lock()
 	defer s.senderReportMu.Unlock()
@@ -527,74 +574,66 @@ func (s *StreamTrackerManager) SetRTCPSenderReportData(layer int32, srFirst *buf
 
 	s.senderReports[layer].first = srFirst
 	s.senderReports[layer].newest = srNewest
+	s.senderReports[layer].lastUpdated = time.Now()
+
+	// (re)fill offsets as necessary for received layer.
+	for i := int32(0); i < buffer.DefaultMaxLayerSpatial+1; i++ {
+		if i == layer {
+			continue
+		}
+
+		// treating layer for which report was received as reference layer
+		s.updateLayerOffsetLocked(layer, i)
+
+		// and the other way
+		s.updateLayerOffsetLocked(i, layer)
+	}
 }
 
-func (s *StreamTrackerManager) GetRTCPSenderReportData(layer int32) (*buffer.RTCPSenderReportData, *buffer.RTCPSenderReportData) {
+func (s *StreamTrackerManager) GetCalculatedClockRate(layer int32) uint32 {
 	s.senderReportMu.RLock()
 	defer s.senderReportMu.RUnlock()
 
 	if layer < 0 || int(layer) >= len(s.senderReports) {
-		return nil, nil
+		// invalid layer
+		return 0
 	}
 
-	return s.senderReports[layer].first, s.senderReports[layer].newest
+	srFirst := s.senderReports[layer].first
+	srNewest := s.senderReports[layer].newest
+	if srFirst == nil || srFirst.NTPTimestamp == 0 || srNewest == nil || srNewest.NTPTimestamp == 0 || srFirst.RTPTimestamp == srNewest.RTPTimestamp {
+		// sender reports invalid or same
+		return 0
+	}
+
+	if s.senderReports[layer].lastUpdated.IsZero() || time.Since(s.senderReports[layer].lastUpdated).Seconds() > senderReportThresholdSeconds {
+		// sender report updated too far back
+		return 0
+	}
+
+	tsf := srNewest.NTPTimestamp.Time().Sub(srFirst.NTPTimestamp.Time())
+	if tsf < minDurationForClockRateCalculation {
+		// not enough time has elapsed to get a stable clock rate calculation
+		return 0
+	}
+
+	rdsf := srNewest.RTPTimestampExt - srFirst.RTPTimestampExt
+	return uint32(float64(rdsf) / tsf.Seconds())
 }
 
 func (s *StreamTrackerManager) GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) {
 	s.senderReportMu.RLock()
 	defer s.senderReportMu.RUnlock()
 
-	if layer < 0 || referenceLayer < 0 {
+	if layer < 0 || int(layer) >= len(s.layerOffsets[0]) || referenceLayer < 0 || int(referenceLayer) >= len(s.layerOffsets) {
 		return 0, fmt.Errorf("invalid layer, target: %d, reference: %d", layer, referenceLayer)
 	}
 
-	if layer == referenceLayer {
-		return ts, nil
+	if layer != referenceLayer && s.layerOffsets[referenceLayer][layer] == 0 {
+		return 0, fmt.Errorf("offset unavailable, target: %d, reference: %d", layer, referenceLayer)
 	}
 
-	var srLayer *buffer.RTCPSenderReportData
-	if int(layer) < len(s.senderReports) {
-		srLayer = s.senderReports[layer].newest
-	}
-	if srLayer == nil || srLayer.NTPTimestamp == 0 {
-		return 0, fmt.Errorf("layer rtcp sender report not available: %d", layer)
-	}
-
-	var srRef *buffer.RTCPSenderReportData
-	if int(referenceLayer) < len(s.senderReports) {
-		srRef = s.senderReports[referenceLayer].newest
-	}
-	if srRef == nil || srRef.NTPTimestamp == 0 {
-		return 0, fmt.Errorf("reference layer rtcp sender report not available: %d", referenceLayer)
-	}
-
-	// line up the RTP time stamps using NTP time of most recent sender report of layer and referenceLayer
-	// NOTE: It is possible that reference layer has stopped (due to dynacast/adaptive streaming OR publisher
-	// constraints). It should be okay even if the layer has stopped for a long time when using modulo arithmetic for
-	// RTP time stamp (uint32 arithmetic).
-	ntpDiff := srRef.NTPTimestamp.Time().Sub(srLayer.NTPTimestamp.Time())
-	rtpDiff := ntpDiff.Nanoseconds() * int64(s.clockRate) / 1e9
-	normalizedTS := srLayer.RTPTimestamp + uint32(rtpDiff)
-	s.logger.Infow(
-		"getting reference timestamp",
-		"layer", layer,
-		"referenceLayer", referenceLayer,
-		"incomingTS", ts,
-		"layerNTP", srLayer.NTPTimestamp.Time().String(),
-		"refNTP", srRef.NTPTimestamp.Time().String(),
-		"ntpDiff", ntpDiff.String(),
-		"layerRTP", srLayer.RTPTimestamp,
-		"refRTP", srRef.RTPTimestamp,
-		"rtpDiff", rtpDiff,
-		"normalizedTS", normalizedTS,
-		"mappedTS", ts+(srRef.RTPTimestamp-normalizedTS),
-	)
-
-	// now that both RTP timestamps correspond to roughly the same NTP time,
-	// the diff between them is the offset in RTP timestamp units between layer and referenceLayer.
-	// Add the offset to layer's ts to map it to corresponding RTP timestamp in
-	// the reference layer.
-	return ts + (srRef.RTPTimestamp - normalizedTS), nil
+	return ts + s.layerOffsets[referenceLayer][layer], nil
 }
 
 func (s *StreamTrackerManager) GetMaxTemporalLayerSeen() int32 {

pkg/sfu/videolayerselector/dependencydescriptor.go

@@ -204,26 +204,6 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r
 
 		d.previousActiveDecodeTargetsBitmask = d.activeDecodeTargetsBitmask
 		d.activeDecodeTargetsBitmask = buffer.GetActiveDecodeTargetBitmask(d.currentLayer, ddwdt.DecodeTargets)
-
-		if d.currentLayer.Spatial == d.requestSpatial {
-			result.IsSwitchingToRequestSpatial = true
-		}
-		if d.currentLayer.Spatial == d.maxLayer.Spatial {
-			result.IsSwitchingToMaxSpatial = true
-			result.MaxSpatialLayer = d.currentLayer.Spatial
-			d.logger.Infow(
-				"reached max layer",
-				"previous", d.previousLayer,
-				"current", d.currentLayer,
-				"previousTarget", d.previousTargetLayer,
-				"target", d.targetLayer,
-				"max", d.maxLayer,
-				"layer", fd.SpatialId,
-				"req", d.requestSpatial,
-				"maxSeen", d.maxSeenLayer,
-				"feed", extPkt.Packet.SSRC,
-			)
-		}
 	}
 
 	ddExtension := &dede.DependencyDescriptorExtension{

pkg/sfu/videolayerselector/simulcast.go

@@ -26,25 +26,15 @@ func (s *Simulcast) IsOvershootOkay() bool {
 }
 
 func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoLayerSelectorResult) {
-	populateSwitches := func(isActive bool, reason string) {
-		result.IsSwitching = true
+	populateSwitches := func(isSwitching bool, isActive bool, reason string) {
+		if isSwitching {
+			result.IsSwitching = true
+		}
+
 		if !isActive {
 			result.IsResuming = true
 		}
 
-		if s.currentLayer.Spatial == s.requestSpatial {
-			result.IsSwitchingToRequestSpatial = true
-		}
-
-		if s.currentLayer.Spatial >= s.maxLayer.Spatial {
-			result.IsSwitchingToMaxSpatial = true
-			result.MaxSpatialLayer = s.currentLayer.Spatial
-			if reason != "" {
-				reason += ", "
-			}
-			reason += "reached max layer"
-		}
-
 		if reason != "" {
 			s.logger.Infow(
 				reason,
@@ -70,6 +60,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL
 		//   1. Resumable layer - don't need a key frame
 		//   2. Opportunistic layer upgrade - needs a key frame
 		//   3. Need to downgrade - needs a key frame
+		isSwitching := true
 		isActive := s.currentLayer.IsValid()
 		found := false
 		reason := ""
@@ -77,6 +68,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL
 			if s.parkedLayer.Spatial == layer {
 				reason = "resuming at parked layer"
 				currentLayer = s.parkedLayer
+				isSwitching = false
 				found = true
 			}
 		} else {
@@ -110,7 +102,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL
 				s.targetLayer.Spatial = s.currentLayer.Spatial
 			}
 
-			populateSwitches(isActive, reason)
+			populateSwitches(isSwitching, isActive, reason)
 		}
 	}
 
@@ -124,7 +116,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL
 			s.targetLayer.Spatial = layer
 		}
 
-		populateSwitches(true, "adjusting overshoot")
+		populateSwitches(true, true, "adjusting overshoot")
 	}
 
 	result.RTPMarker = extPkt.Packet.Marker

pkg/sfu/videolayerselector/videolayerselector.go

@@ -10,9 +10,6 @@ type VideoLayerSelectorResult struct {
 	IsRelevant                    bool
 	IsSwitching                   bool
 	IsResuming                    bool
-	IsSwitchingToRequestSpatial   bool
-	IsSwitchingToMaxSpatial       bool
-	MaxSpatialLayer               int32
 	RTPMarker                     bool
 	DependencyDescriptorExtension []byte
 }

pkg/sfu/videolayerselector/vp9.go

@@ -80,26 +80,6 @@ func (v *VP9) Select(extPkt *buffer.ExtPacket, _layer int32) (result VideoLayerS
 				result.IsResuming = true
 			}
 
-			if v.currentLayer.Spatial != v.requestSpatial && updatedLayer.Spatial == v.requestSpatial {
-				result.IsSwitchingToRequestSpatial = true
-			}
-
-			if v.currentLayer.Spatial != v.maxLayer.Spatial && updatedLayer.Spatial == v.maxLayer.Spatial {
-				result.IsSwitchingToMaxSpatial = true
-				result.MaxSpatialLayer = updatedLayer.Spatial
-				v.logger.Infow(
-					"reached max layer",
-					"current", v.currentLayer,
-					"updated", updatedLayer,
-					"target", v.targetLayer,
-					"max", v.maxLayer,
-					"layer", extPkt.VideoLayer.Spatial,
-					"req", v.requestSpatial,
-					"maxSeen", v.maxSeenLayer,
-					"feed", extPkt.Packet.SSRC,
-				)
-			}
-
 			v.previousLayer = v.currentLayer
 			v.currentLayer = updatedLayer
 		}