FPS (#410)

* WIP commit

* WIP commit

* WIP commit

* WIP commit

* WIP commit

* WIP commit

* Clean up

* Clean up

* Store RTT in stats

* spelling mistake

* Make tests compile

* Fix test compilation error

* fix tests

* clone

* latest protocol

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/google/wire v0.5.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/hashicorp/golang-lru v0.5.4
-	github.com/livekit/protocol v0.11.13-0.20220207221412-40df25091bcb
+	github.com/livekit/protocol v0.11.13-0.20220208070440-7f8399010124
 	github.com/magefile/mage v1.11.0
 	github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
 	github.com/mitchellh/go-homedir v1.1.0

go.sum

@@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA=
 github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
-github.com/livekit/protocol v0.11.13-0.20220207221412-40df25091bcb h1:oAx83edonLEOfsZardevXlHqzut6dEBIH9fOU/ZxDfU=
-github.com/livekit/protocol v0.11.13-0.20220207221412-40df25091bcb/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
+github.com/livekit/protocol v0.11.13-0.20220208070440-7f8399010124 h1:nZor+yFxIBKLdQGwwx6GvsnjZhmG9gcz9I2K2zdLiVI=
+github.com/livekit/protocol v0.11.13-0.20220208070440-7f8399010124/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
 github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls=
 github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
 github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=

pkg/rtc/mediatrack.go

@@ -6,8 +6,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
-
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/pion/rtcp"
@@ -36,8 +34,6 @@ type MediaTrack struct {
 	audioLevelMu sync.RWMutex
 	audioLevel   *AudioLevel
 
-	connectionStats *connectionquality.ConnectionStats
-
 	done chan struct{}
 
 	*MediaTrackReceiver
@@ -97,30 +93,6 @@ func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTra
 		}
 	})
 
-	t.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{
-		UpdateInterval: connectionQualityUpdateInterval,
-		CodecType:      track.Kind(),
-		GetTotalBytes: func() uint64 {
-			receiver := t.Receiver()
-			if receiver != nil {
-				return receiver.(*sfu.WebRTCReceiver).GetTotalBytes()
-			}
-
-			return 0
-		},
-		GetIsReducedQuality: func() bool {
-			publishing, expected := t.getNumUpTracks()
-			return publishing < expected
-		},
-		Logger: t.params.Logger,
-	})
-	t.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) {
-		t.params.Telemetry.TrackStats(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), stat)
-	})
-
-	t.AddOnClose(func() {
-		t.connectionStats.Close()
-	})
 	return t
 }
 
@@ -147,23 +119,6 @@ func (t *MediaTrack) ToProto() *livekit.TrackInfo {
 	return info
 }
 
-func (t *MediaTrack) getNumUpTracks() (uint32, uint32) {
-	numExpected := atomic.LoadUint32(&t.numUpTracks)
-
-	numSubscribedLayers := t.numSubscribedLayers()
-	if numSubscribedLayers < numExpected {
-		numExpected = numSubscribedLayers
-	}
-
-	numPublishing := uint32(0)
-	receiver := t.Receiver()
-	if receiver != nil {
-		numPublishing = uint32(receiver.(*sfu.WebRTCReceiver).NumAvailableSpatialLayers())
-	}
-
-	return numPublishing, numExpected
-}
-
 // AddReceiver adds a new RTP receiver to the track
 func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, twcc *twcc.Responder) {
 	buff, rtcpReader := t.params.BufferFactory.GetBufferPair(uint32(track.SSRC()))
@@ -171,7 +126,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 		t.params.Logger.Errorw("could not retrieve buffer pair", nil)
 		return
 	}
-	buff.OnFeedback(t.handlePublisherFeedback)
 
 	if t.Kind() == livekit.TrackType_AUDIO {
 		t.audioLevelMu.Lock()
@@ -225,13 +179,14 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 			t.MediaTrackReceiver.Close()
 			t.params.Telemetry.TrackUnpublished(context.Background(), t.PublisherID(), t.ToProto(), uint32(track.SSRC()))
 		})
+		wr.OnStatsUpdate(func(_ *sfu.WebRTCReceiver, stat *livekit.AnalyticsStat) {
+			t.params.Telemetry.TrackStats(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), stat)
+		})
 		t.params.Telemetry.TrackPublished(context.Background(), t.PublisherID(), t.ToProto())
 
 		t.buffer = buff
 
 		t.MediaTrackReceiver.SetupReceiver(wr)
-
-		t.connectionStats.Start()
 	}
 	t.lock.Unlock()
 
@@ -272,14 +227,11 @@ func (t *MediaTrack) GetAudioLevel() (level uint8, active bool) {
 	return t.audioLevel.GetLevel()
 }
 
-func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) {
-	t.connectionStats.RTCPFeedback(packets, 0)
+func (t *MediaTrack) GetConnectionScore() float32 {
+	receiver := t.Receiver()
+	if receiver == nil {
+		return 0.0
+	}
 
-	// also look for sender reports
-	// feedback for the source RTCP
-	t.params.RTCPChan <- packets
-}
-
-func (t *MediaTrack) GetConnectionScore() float64 {
-	return t.connectionStats.GetScore()
+	return receiver.(*sfu.WebRTCReceiver).GetConnectionScore()
 }

pkg/rtc/mediatrackreceiver.go

@@ -291,7 +291,7 @@ func (t *MediaTrackReceiver) GetQualityForDimension(width, height uint32) liveki
 	return quality
 }
 
-// handles max loss for audio packets
+// handles max loss for audio streams
 func (t *MediaTrackReceiver) handleMaxLossFeedback(_ *sfu.DownTrack, report *rtcp.ReceiverReport) {
 	t.downFracLostLock.Lock()
 	for _, rr := range report.Reports {

pkg/rtc/participant.go

@@ -762,9 +762,9 @@ func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo
 	}
 	p.lock.RUnlock()
 
-	avgScore := 5.0
+	avgScore := float32(5.0)
 	if numTracks > 0 {
-		avgScore = totalScore / float64(numTracks)
+		avgScore = totalScore / float32(numTracks)
 	}
 
 	rating := connectionquality.Score2Rating(avgScore)
@@ -1348,7 +1348,7 @@ func (p *ParticipantImpl) setTrackMuted(trackID livekit.TrackID, muted bool) {
 	}
 }
 
-func (p *ParticipantImpl) getPublisherConnectionQuality() (totalScore float64, numTracks int) {
+func (p *ParticipantImpl) getPublisherConnectionQuality() (totalScore float32, numTracks int) {
 	for _, pt := range p.GetPublishedTracks() {
 		if pt.IsMuted() {
 			continue

pkg/rtc/participant_internal_test.go

@@ -244,7 +244,7 @@ func TestMuteSetting(t *testing.T) {
 func TestConnectionQuality(t *testing.T) {
 
 	// loss based score is currently a publisher method.
-	videoScore := func(loss float64, numPublishing, numRegistered uint32) float64 {
+	videoScore := func(loss float32, numPublishing, numRegistered uint32) float32 {
 		var reducedQuality bool
 		if numRegistered > 0 && numPublishing != numRegistered {
 			reducedQuality = true
@@ -252,7 +252,7 @@ func TestConnectionQuality(t *testing.T) {
 		return connectionquality.VideoConnectionScore(loss, reducedQuality)
 	}
 
-	testPublishedVideoTrack := func(loss float64, numPublishing, numRegistered uint32) *typesfakes.FakeLocalMediaTrack {
+	testPublishedVideoTrack := func(loss float32, numPublishing, numRegistered uint32) *typesfakes.FakeLocalMediaTrack {
 		tr := &typesfakes.FakeLocalMediaTrack{}
 		score := videoScore(loss, numPublishing, numRegistered)
 		t.Log("video score: ", score)
@@ -263,11 +263,11 @@ func TestConnectionQuality(t *testing.T) {
 	testPublishedAudioTrack := func(totalPackets, packetsLost uint32) *typesfakes.FakeLocalMediaTrack {
 		tr := &typesfakes.FakeLocalMediaTrack{}
 
-		pctLoss := float64(0.0)
+		pctLoss := float32(0.0)
 		if totalPackets > 0 {
-			pctLoss = (float64(packetsLost) / float64(totalPackets)) * 100.0
+			pctLoss = (float32(packetsLost) / float32(totalPackets)) * 100.0
 		}
-		score := connectionquality.AudioConnectionScore(pctLoss, 0)
+		score := connectionquality.AudioConnectionScore(pctLoss, 0, 0.0)
 		t.Log("audio score: ", score)
 		tr.GetConnectionScoreReturns(score)
 		return tr

pkg/rtc/types/interfaces.go

@@ -211,7 +211,7 @@ type LocalMediaTrack interface {
 	SdpCid() string
 
 	GetAudioLevel() (level uint8, active bool)
-	GetConnectionScore() float64
+	GetConnectionScore() float32
 }
 
 // MediaTrack is the main interface representing a track published to the room

pkg/rtc/types/typesfakes/fake_local_media_track.go

@@ -38,15 +38,15 @@ type FakeLocalMediaTrack struct {
 		result1 uint8
 		result2 bool
 	}
-	GetConnectionScoreStub        func() float64
+	GetConnectionScoreStub        func() float32
 	getConnectionScoreMutex       sync.RWMutex
 	getConnectionScoreArgsForCall []struct {
 	}
 	getConnectionScoreReturns struct {
-		result1 float64
+		result1 float32
 	}
 	getConnectionScoreReturnsOnCall map[int]struct {
-		result1 float64
+		result1 float32
 	}
 	GetQualityForDimensionStub        func(uint32, uint32) livekit.VideoQuality
 	getQualityForDimensionMutex       sync.RWMutex
@@ -387,7 +387,7 @@ func (fake *FakeLocalMediaTrack) GetAudioLevelReturnsOnCall(i int, result1 uint8
 	}{result1, result2}
 }
 
-func (fake *FakeLocalMediaTrack) GetConnectionScore() float64 {
+func (fake *FakeLocalMediaTrack) GetConnectionScore() float32 {
 	fake.getConnectionScoreMutex.Lock()
 	ret, specificReturn := fake.getConnectionScoreReturnsOnCall[len(fake.getConnectionScoreArgsForCall)]
 	fake.getConnectionScoreArgsForCall = append(fake.getConnectionScoreArgsForCall, struct {
@@ -411,32 +411,32 @@ func (fake *FakeLocalMediaTrack) GetConnectionScoreCallCount() int {
 	return len(fake.getConnectionScoreArgsForCall)
 }
 
-func (fake *FakeLocalMediaTrack) GetConnectionScoreCalls(stub func() float64) {
+func (fake *FakeLocalMediaTrack) GetConnectionScoreCalls(stub func() float32) {
 	fake.getConnectionScoreMutex.Lock()
 	defer fake.getConnectionScoreMutex.Unlock()
 	fake.GetConnectionScoreStub = stub
 }
 
-func (fake *FakeLocalMediaTrack) GetConnectionScoreReturns(result1 float64) {
+func (fake *FakeLocalMediaTrack) GetConnectionScoreReturns(result1 float32) {
 	fake.getConnectionScoreMutex.Lock()
 	defer fake.getConnectionScoreMutex.Unlock()
 	fake.GetConnectionScoreStub = nil
 	fake.getConnectionScoreReturns = struct {
-		result1 float64
+		result1 float32
 	}{result1}
 }
 
-func (fake *FakeLocalMediaTrack) GetConnectionScoreReturnsOnCall(i int, result1 float64) {
+func (fake *FakeLocalMediaTrack) GetConnectionScoreReturnsOnCall(i int, result1 float32) {
 	fake.getConnectionScoreMutex.Lock()
 	defer fake.getConnectionScoreMutex.Unlock()
 	fake.GetConnectionScoreStub = nil
 	if fake.getConnectionScoreReturnsOnCall == nil {
 		fake.getConnectionScoreReturnsOnCall = make(map[int]struct {
-			result1 float64
+			result1 float32
 		})
 	}
 	fake.getConnectionScoreReturnsOnCall[i] = struct {
-		result1 float64
+		result1 float32
 	}{result1}
 }
 

pkg/sfu/buffer/buffer.go

@@ -3,6 +3,7 @@ package buffer
 import (
 	"encoding/binary"
 	"io"
+	"math/rand"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -36,7 +37,7 @@ type ExtPacket struct {
 
 // Buffer contains all packets
 type Buffer struct {
-	sync.Mutex
+	sync.RWMutex
 	bucket     *Bucket
 	nacker     *NackQueue
 	videoPool  *sync.Pool
@@ -63,21 +64,20 @@ type Buffer struct {
 	latestTSForAudioLevelInitialized bool
 	latestTSForAudioLevel            uint32
 
-	lastPacketRead     int
-	bitrate            atomic.Value
-	bitrateHelper      [4]int64
-	lastSRNTPTime      uint64
-	lastSRRTPTime      uint32
-	lastSRRecv         int64 // Represents wall clock of the most recent sender report arrival
-	cycle              uint16
-	lastRtcpPacketTime int64 // Time the last RTCP packet was received.
-	lastRtcpSrTime     int64 // Time the last RTCP SR was received. Required for DLSR computation.
-	lastTransit        uint32
+	lastPacketRead int
+	bitrate        atomic.Value
+	bitrateHelper  [4]int64
+	lastSRNTPTime  uint64
+	lastSRRTPTime  uint32
+	lastSRRecv     int64 // Represents wall clock of the most recent sender report arrival
+	lastTransit    uint32
 
-	stats      Stats
+	started    bool
+	stats      StreamStats
 	rrSnapshot *receiverReportSnapshot
 
 	highestSN uint16
+	cycle     uint16
 
 	lastFractionLostToReport uint8 // Last fraction lost from subscribers, should report to publisher; Audio only
 
@@ -91,17 +91,10 @@ type Buffer struct {
 	logger logger.Logger
 }
 
-type Stats struct {
-	PacketCount uint32 // Number of packets received from this source.
-	TotalBytes  uint64
-	Jitter      float64 // An estimate of the statistical variance of the RTP data packet inter-arrival time.
-}
-
 type receiverReportSnapshot struct {
-	extSeqNum       uint32
-	packetsReceived uint32
-	packetsLost     uint32
-	lastLossRate    float32
+	extHighestSeqNum uint32
+	packetsLost      uint32
+	lastLossRate     float32
 }
 
 // BufferOptions provides configuration options for the buffer
@@ -164,7 +157,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 			case webrtc.TypeRTCPFBNACK:
 				b.logger.Debugw("Setting feedback", "type", webrtc.TypeRTCPFBNACK)
 				b.nacker = NewNACKQueue()
-				b.nacker.SetRTT(70) // LK-TODO: sane default till we get better data
+				b.nacker.SetRTT(70) // default till it is updated
 				b.nack = true
 			}
 		}
@@ -271,6 +264,29 @@ func (b *Buffer) OnClose(fn func()) {
 	b.onClose = fn
 }
 
+func (b *Buffer) SendPLI() {
+	b.Lock()
+	b.stats.TotalPLIs++
+	b.Unlock()
+
+	pli := []rtcp.Packet{
+		&rtcp.PictureLossIndication{SenderSSRC: rand.Uint32(), MediaSSRC: b.mediaSSRC},
+	}
+
+	b.feedbackCB(pli)
+}
+
+func (b *Buffer) SetRTT(rtt uint32) {
+	b.Lock()
+	defer b.Unlock()
+
+	b.stats.RTT = rtt
+
+	if b.nacker != nil {
+		b.nacker.SetRTT(rtt)
+	}
+}
+
 func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
 	isRTX := false
 
@@ -328,41 +344,58 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
 func (b *Buffer) updateStreamState(p *rtp.Packet, pktSize int, arrivalTime int64, isRTX bool) {
 	sn := p.SequenceNumber
 
-	if b.stats.PacketCount == 0 {
-		b.highestSN = sn - 1
+	if !b.started {
+		b.started = true
+		b.highestSN = sn
 
 		b.lastReport = arrivalTime
 
 		b.rrSnapshot = &receiverReportSnapshot{
-			extSeqNum:       uint32(sn) - 1,
-			packetsReceived: 0,
-			packetsLost:     0,
-			lastLossRate:    0.0,
-		}
-	}
-
-	diff := sn - b.highestSN
-	if diff > (1 << 15) {
-		// out-of-order, remove it from nack queue
-		if b.nacker != nil {
-			b.nacker.Remove(sn)
+			extHighestSeqNum: uint32(sn) - 1,
+			packetsLost:      0,
+			lastLossRate:     0.0,
 		}
 	} else {
-		if b.nacker != nil && diff > 1 {
-			for lost := b.highestSN + 1; lost != sn; lost++ {
-				b.nacker.Push(lost)
+		diff := sn - b.highestSN
+		if diff > (1 << 15) {
+			if !isRTX && b.stats.TotalPacketsLost != 0 {
+				b.stats.TotalPacketsLost--
 			}
-		}
 
-		if sn < b.highestSN && b.stats.PacketCount > 0 {
-			b.cycle++
-		}
+			// out-of-order, remove it from nack queue
+			if b.nacker != nil {
+				b.nacker.Remove(sn)
+			}
+		} else {
+			b.stats.TotalPacketsLost += (uint32(diff) - 1)
+			if b.nacker != nil && diff > 1 {
+				for lost := b.highestSN + 1; lost != sn; lost++ {
+					b.nacker.Push(lost)
+				}
+			}
 
-		b.highestSN = sn
+			if sn < b.highestSN {
+				b.cycle++
+			}
+
+			b.highestSN = sn
+		}
 	}
 
-	b.stats.PacketCount++
-	b.stats.TotalBytes += uint64(pktSize)
+	switch {
+	case isRTX:
+		b.stats.TotalRetransmitPackets++
+		b.stats.TotalRetransmitBytes += uint64(pktSize)
+	case len(p.Payload) == 0:
+		b.stats.TotalPaddingPackets++
+		b.stats.TotalPaddingBytes += uint64(pktSize)
+	default:
+		b.stats.TotalPrimaryPackets++
+		b.stats.TotalPrimaryBytes += uint64(pktSize)
+		if p.Marker {
+			b.stats.TotalFrames++
+		}
+	}
 
 	if !isRTX {
 		// jitter
@@ -445,8 +478,9 @@ func (b *Buffer) doNACKs() {
 		return
 	}
 
-	if r := b.buildNACKPacket(); r != nil {
-		b.feedbackCB(r)
+	if r, numSeqNumsNacked := b.buildNACKPacket(); r != nil {
+		go b.feedbackCB(r)
+		b.stats.TotalNACKs += uint32(numSeqNumsNacked)
 	}
 }
 
@@ -476,11 +510,11 @@ func (b *Buffer) doReports(arrivalTime int64) {
 	b.bitrate.Store(bitrates)
 
 	// RTCP reports
-	b.feedbackCB(b.getRTCP())
+	go b.feedbackCB(b.getRTCP())
 }
 
-func (b *Buffer) buildNACKPacket() []rtcp.Packet {
-	if nacks := b.nacker.Pairs(); len(nacks) > 0 {
+func (b *Buffer) buildNACKPacket() ([]rtcp.Packet, int) {
+	if nacks, numSeqNumsNacked := b.nacker.Pairs(); len(nacks) > 0 {
 		var pkts []rtcp.Packet
 		if len(nacks) > 0 {
 			pkts = []rtcp.Packet{&rtcp.TransportLayerNack{
@@ -489,9 +523,9 @@ func (b *Buffer) buildNACKPacket() []rtcp.Packet {
 			}}
 		}
 
-		return pkts
+		return pkts, numSeqNumsNacked
 	}
-	return nil
+	return nil, 0
 }
 
 func (b *Buffer) buildREMBPacket() *rtcp.ReceiverEstimatedMaximumBitrate {
@@ -514,7 +548,6 @@ func (b *Buffer) buildREMBPacket() *rtcp.ReceiverEstimatedMaximumBitrate {
 	if br < 100000 {
 		br = 100000
 	}
-	b.stats.TotalBytes = 0
 
 	return &rtcp.ReceiverEstimatedMaximumBitrate{
 		Bitrate: float32(br),
@@ -527,14 +560,13 @@ func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport {
 		return nil
 	}
 
-	extMaxSeq := (uint32(b.cycle) << 16) | uint32(b.highestSN)
-	expectedInInterval := extMaxSeq - b.rrSnapshot.extSeqNum
+	extHighestSeqNum := (uint32(b.cycle) << 16) | uint32(b.highestSN)
+	expectedInInterval := extHighestSeqNum - b.rrSnapshot.extHighestSeqNum
 	if expectedInInterval == 0 {
 		return nil
 	}
 
-	receivedInInterval := b.stats.PacketCount - b.rrSnapshot.packetsReceived
-	lostInInterval := expectedInInterval - receivedInInterval
+	lostInInterval := b.stats.TotalPacketsLost - b.rrSnapshot.packetsLost
 	if int32(lostInInterval) < 0 {
 		// could happen if retransmitted packets arrive and make received greater than expected
 		lostInInterval = 0
@@ -547,8 +579,6 @@ func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport {
 		fracLost = b.lastFractionLostToReport
 	}
 
-	totalLost := b.rrSnapshot.packetsLost + lostInInterval
-
 	var dlsr uint32
 	if b.lastSRRecv != 0 {
 		delayMS := uint32((time.Now().UnixNano() - b.lastSRRecv) / 1e6)
@@ -557,17 +587,16 @@ func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport {
 	}
 
 	b.rrSnapshot = &receiverReportSnapshot{
-		extSeqNum:       extMaxSeq,
-		packetsReceived: b.stats.PacketCount,
-		packetsLost:     totalLost,
-		lastLossRate:    lossRate,
+		extHighestSeqNum: extHighestSeqNum,
+		packetsLost:      b.stats.TotalPacketsLost,
+		lastLossRate:     lossRate,
 	}
 
 	return &rtcp.ReceptionReport{
 		SSRC:               b.mediaSSRC,
 		FractionLost:       fracLost,
-		TotalLost:          totalLost,
-		LastSequenceNumber: extMaxSeq,
+		TotalLost:          b.stats.TotalPacketsLost,
+		LastSequenceNumber: extHighestSeqNum,
 		Jitter:             uint32(b.stats.Jitter),
 		LastSenderReport:   uint32(b.lastSRNTPTime >> 16),
 		Delay:              dlsr,
@@ -677,22 +706,19 @@ func (b *Buffer) GetSenderReportData() (rtpTime uint32, ntpTime uint64, lastRece
 	return rtpTime, ntpTime, lastReceivedTimeInNanosSinceEpoch
 }
 
-// GetStats returns the raw statistics about a particular buffer state
-func (b *Buffer) GetStats() (stats Stats) {
-	b.Lock()
-	stats = b.stats
-	b.Unlock()
-	return
-}
+func (b *Buffer) GetStats() *StreamStatsWithLayers {
+	b.RLock()
+	defer b.RUnlock()
 
-// Used only in tests
-func (b *Buffer) SetStatsTestOnly(stats Stats) {
-	b.Lock()
-	b.stats = stats
-	b.Unlock()
-}
+	layers := make(map[int]LayerStats)
+	layers[0] = LayerStats{
+		TotalPackets: b.stats.TotalPrimaryPackets + b.stats.TotalRetransmitPackets + b.stats.TotalPaddingPackets,
+		TotalBytes:   b.stats.TotalPrimaryBytes + b.stats.TotalRetransmitBytes + b.stats.TotalPaddingBytes,
+		TotalFrames:  b.stats.TotalFrames,
+	}
 
-// IsLaterTimestamp returns true if timestamp1 is later in time than timestamp2 factoring in timestamp wrap-around
-func IsLaterTimestamp(timestamp1 uint32, timestamp2 uint32) bool {
-	return (timestamp1 - timestamp2) < (1 << 31)
+	return &StreamStatsWithLayers{
+		StreamStats: b.stats,
+		Layers:      layers,
+	}
 }

pkg/sfu/buffer/buffer_test.go

@@ -244,59 +244,3 @@ func TestFractionLostReport(t *testing.T) {
 	}
 	wg.Wait()
 }
-
-/*
-func TestSeqWrapHandler(t *testing.T) {
-	s := SeqWrapHandler{}
-	s.UpdateMaxSeq(1)
-	require.Equal(t, uint32(1), s.MaxSeqNo())
-
-	type caseInfo struct {
-		seqs  []uint32 // {seq1, seq2, unwrap of seq2}
-		newer bool     // seq2 is newer than seq1
-	}
-	// test normal case, name -> {seq1, seq2, unwrap of seq2}
-	cases := map[string]caseInfo{
-		"no wrap":                             {[]uint32{1, 4, 4}, true},
-		"no wrap backward":                    {[]uint32{4, 1, 1}, false},
-		"wrap around forward to zero":         {[]uint32{65534, 0, 65536}, true},
-		"wrap around forward":                 {[]uint32{65534, 10, 65546}, true},
-		"wrap around forward 2":               {[]uint32{65535 + 65536*2, 1, 1 + 65536*3}, true},
-		"wrap around backward ":               {[]uint32{5, 65534, 65534}, false},
-		"wrap around backward less than zero": {[]uint32{5, 65534, 65534}, false},
-	}
-
-	for k, v := range cases {
-		t.Run(k, func(t *testing.T) {
-			s := SeqWrapHandler{}
-			s.UpdateMaxSeq(v.seqs[0])
-			extsn, newer := s.Unwrap(uint16(v.seqs[1]))
-			require.Equal(t, v.newer, newer)
-			require.Equal(t, v.seqs[2], extsn)
-		})
-	}
-
-}
-
-func TestIsTimestampWrap(t *testing.T) {
-	type caseInfo struct {
-		name  string
-		ts1   uint32
-		ts2   uint32
-		later bool
-	}
-
-	cases := []caseInfo{
-		{"normal case 1 timestamp later ", 2, 1, true},
-		{"normal case 2 timestamp later", 0x1c000000, 0x10000000, true},
-		{"wrap case timestamp later", 0xffff, 0xfc000000, true},
-		{"wrap case timestamp early", 0xfc000000, 0xffff, false},
-	}
-
-	for _, c := range cases {
-		t.Run(c.name, func(t *testing.T) {
-			require.Equal(t, c.later, IsLaterTimestamp(c.ts1, c.ts2))
-		})
-	}
-}
-*/

pkg/sfu/buffer/nack.go

@@ -26,7 +26,7 @@ func NewNACKQueue() *NackQueue {
 	}
 }
 
-func (n *NackQueue) SetRTT(rtt int) {
+func (n *NackQueue) SetRTT(rtt uint32) {
 	n.rtt = time.Duration(rtt) * time.Millisecond
 }
 
@@ -52,9 +52,9 @@ func (n *NackQueue) Push(sn uint16) {
 	n.nacks = append(n.nacks, &nack{seqNum: sn, nacked: 0, lastNackTime: time.Now()})
 }
 
-func (n *NackQueue) Pairs() []rtcp.NackPair {
+func (n *NackQueue) Pairs() ([]rtcp.NackPair, int) {
 	if len(n.nacks) == 0 {
-		return nil
+		return nil, 0
 	}
 
 	now := time.Now()
@@ -64,6 +64,7 @@ func (n *NackQueue) Pairs() []rtcp.NackPair {
 
 	snsToPurge := []uint16{}
 
+	numSeqNumsNacked := 0
 	isPairActive := false
 	var np rtcp.NackPair
 	var nps []rtcp.NackPair
@@ -77,6 +78,7 @@ func (n *NackQueue) Pairs() []rtcp.NackPair {
 
 		nack.nacked++
 		nack.lastNackTime = now
+		numSeqNumsNacked++
 
 		if (nack.seqNum - baseSN) > 16 {
 			// need a new nack pair
@@ -104,5 +106,5 @@ func (n *NackQueue) Pairs() []rtcp.NackPair {
 		n.Remove(sn)
 	}
 
-	return nps
+	return nps, numSeqNumsNacked
 }

pkg/sfu/buffer/nack_test.go

@@ -8,44 +8,61 @@ import (
 )
 
 func Test_nackQueue_pairs(t *testing.T) {
+	type PairsResult struct {
+		pairs            []rtcp.NackPair
+		numSeqNumsNacked int
+	}
+
 	tests := []struct {
 		name string
 		args []uint16
-		want []rtcp.NackPair
+		want PairsResult
 	}{
 		{
 			name: "Must return correct single pairs pair",
 			args: []uint16{1, 2, 4, 5},
-			want: []rtcp.NackPair{{
-				PacketID:    1,
-				LostPackets: 13,
-			}},
+			want: PairsResult{
+				pairs: []rtcp.NackPair{
+					{
+						PacketID:    1,
+						LostPackets: 13,
+					},
+				},
+				numSeqNumsNacked: 4,
+			},
 		},
 		{
 			name: "Must return correct pair wrap",
 			args: []uint16{65533, 2, 4, 5, 30, 32},
-			want: []rtcp.NackPair{
-				{
-					PacketID:    65533,
-					LostPackets: 1<<7 + 1<<6 + 1<<4,
+			want: PairsResult{
+				pairs: []rtcp.NackPair{
+					{
+						PacketID:    65533,
+						LostPackets: 1<<7 + 1<<6 + 1<<4,
+					},
+					{
+						PacketID:    30,
+						LostPackets: 1 << 1,
+					},
 				},
-				{
-					PacketID:    30,
-					LostPackets: 1 << 1,
-				}},
+				numSeqNumsNacked: 6,
+			},
 		},
 		{
 			name: "Must return 2 pairs pair",
 			args: []uint16{1, 2, 4, 5, 20, 22, 24, 27},
-			want: []rtcp.NackPair{
-				{
-					PacketID:    1,
-					LostPackets: 13,
-				},
-				{
-					PacketID:    20,
-					LostPackets: 74,
+			want: PairsResult{
+				pairs: []rtcp.NackPair{
+					{
+						PacketID:    1,
+						LostPackets: 13,
+					},
+					{
+						PacketID:    20,
+						LostPackets: 74,
+					},
 				},
+				numSeqNumsNacked: 8,
 			},
 		},
 	}
@@ -56,8 +73,9 @@ func Test_nackQueue_pairs(t *testing.T) {
 			for _, sn := range tt.args {
 				n.Push(sn)
 			}
-			got := n.Pairs()
-			require.EqualValues(t, tt.want, got)
+			got, numSeqNumsNacked := n.Pairs()
+			require.EqualValues(t, tt.want.pairs, got)
+			require.Equal(t, tt.want.numSeqNumsNacked, numSeqNumsNacked)
 		})
 	}
 }

pkg/sfu/buffer/streamstats.go

@@ -0,0 +1,28 @@
+package buffer
+
+type LayerStats struct {
+	TotalPackets uint32
+	TotalBytes   uint64
+	TotalFrames  uint32
+}
+
+type StreamStats struct {
+	TotalPrimaryPackets    uint32
+	TotalPrimaryBytes      uint64
+	TotalRetransmitPackets uint32
+	TotalRetransmitBytes   uint64
+	TotalPaddingPackets    uint32
+	TotalPaddingBytes      uint64
+	TotalPacketsLost       uint32
+	TotalFrames            uint32
+	RTT                    uint32
+	Jitter                 float64
+	TotalNACKs             uint32
+	TotalPLIs              uint32
+	TotalFIRs              uint32
+}
+
+type StreamStatsWithLayers struct {
+	StreamStats StreamStats
+	Layers      map[int]LayerStats
+}

pkg/sfu/connectionquality/connectionstats.go

@@ -4,60 +4,54 @@ import (
 	"sync"
 	"time"
 
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/utils"
-	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
 )
 
-type Snapshot struct {
-	Initialized      bool
-	TotalPacketsLost uint32
-	HighestSeqNum    uint32
+const (
+	connectionQualityUpdateInterval = 5 * time.Second
+)
+
+type qualityWindow struct {
+	startSeqNum      uint32
+	endSeqNum        uint32
+	startPacketsLost uint32
+	endPacketsLost   uint32
+	maxRTT           uint32
+	maxJitter        uint32
 }
 
 type ConnectionStatsParams struct {
 	UpdateInterval      time.Duration
 	CodecType           webrtc.RTPCodecType
-	GetTotalBytes       func() uint64
+	ClockRate           uint32
+	GetTrackStats       func() map[uint32]*buffer.StreamStatsWithLayers
 	GetIsReducedQuality func() bool
 	Logger              logger.Logger
 }
 
 type ConnectionStats struct {
-	lock sync.RWMutex
-
 	params ConnectionStatsParams
 
-	baseSeqNumInitialized bool
-	baseSeqNum            uint32
-	highestSeqNum         uint32
-	totalPacketsLost      uint32
-	totalBytes            uint64
-
-	snapshot Snapshot
-
-	maxDelay  uint32
-	maxJitter uint32
-
-	nackCount int32
-	pliCount  int32
-	firCount  int32
-
-	score float64
-
 	onStatsUpdate func(cs *ConnectionStats, stat *livekit.AnalyticsStat)
 
+	lock           sync.RWMutex
+	score          float32
+	qualityWindows map[uint32]*qualityWindow
+
 	done     chan struct{}
 	isClosed utils.AtomicFlag
 }
 
 func NewConnectionStats(params ConnectionStatsParams) *ConnectionStats {
 	return &ConnectionStats{
-		params: params,
-		score:  4.0,
-		done:   make(chan struct{}),
+		params:         params,
+		score:          4.0,
+		qualityWindows: make(map[uint32]*qualityWindow),
+		done:           make(chan struct{}),
 	}
 }
 
@@ -77,7 +71,14 @@ func (cs *ConnectionStats) OnStatsUpdate(fn func(cs *ConnectionStats, stat *live
 	cs.onStatsUpdate = fn
 }
 
-func (cs *ConnectionStats) RTCPFeedback(packets []rtcp.Packet, expectedSSRC uint32) {
+func (cs *ConnectionStats) GetScore() float32 {
+	cs.lock.RLock()
+	defer cs.lock.RUnlock()
+
+	return cs.score
+}
+
+func (cs *ConnectionStats) UpdateWindow(ssrc uint32, extHighestSeqNum uint32, packetsLost uint32, rtt uint32, jitter uint32) {
 	if cs.isClosed.Get() {
 		return
 	}
@@ -85,123 +86,169 @@ func (cs *ConnectionStats) RTCPFeedback(packets []rtcp.Packet, expectedSSRC uint
 	cs.lock.Lock()
 	defer cs.lock.Unlock()
 
-	for _, p := range packets {
-		switch pkt := p.(type) {
-		case *rtcp.ReceiverReport:
-			for _, r := range pkt.Reports {
-				if expectedSSRC != 0 && r.SSRC != expectedSSRC {
-					continue
-				}
+	qw := cs.qualityWindows[ssrc]
+	if qw == nil {
+		qw = &qualityWindow{}
+		cs.qualityWindows[ssrc] = qw
+	}
 
-				if r.Delay > cs.maxDelay {
-					cs.maxDelay = r.Delay
-				}
+	if qw.startSeqNum == 0 {
+		qw.startSeqNum = extHighestSeqNum
+		qw.startPacketsLost = packetsLost
+	}
 
-				if r.Jitter > cs.maxJitter {
-					cs.maxJitter = r.Jitter
-				}
+	if extHighestSeqNum > qw.endSeqNum {
+		qw.endSeqNum = extHighestSeqNum
+		qw.endPacketsLost = packetsLost
+	}
 
-				if !cs.baseSeqNumInitialized {
-					cs.baseSeqNumInitialized = true
-					cs.baseSeqNum = r.LastSequenceNumber
-					cs.highestSeqNum = r.LastSequenceNumber
-					cs.totalPacketsLost = r.TotalLost
-				}
+	if rtt > qw.maxRTT {
+		qw.maxRTT = rtt
+	}
 
-				if r.LastSequenceNumber > cs.highestSeqNum {
-					cs.highestSeqNum = r.LastSequenceNumber
-					cs.totalPacketsLost = r.TotalLost
-				}
-			}
+	if jitter > qw.maxJitter {
+		qw.maxJitter = jitter
+	}
+}
 
-		case *rtcp.TransportLayerNack:
-			nackCount := 0
-			for _, pair := range pkt.Nacks {
-				nackCount += len(pair.PacketList())
-			}
-			cs.nackCount += int32(nackCount)
-
-		case *rtcp.PictureLossIndication:
-			cs.pliCount += 1
-
-		case *rtcp.FullIntraRequest:
-			cs.firCount += 1
+func (cs *ConnectionStats) updateScore() {
+	expectedPacketsInInterval := uint32(0)
+	lostPacketsInInterval := uint32(0)
+	maxRTT := uint32(0)
+	maxJitter := uint32(0)
+	for _, qw := range cs.qualityWindows {
+		expectedPacketsInInterval += qw.endSeqNum - qw.endSeqNum + 1
+		lostPacketsInInterval += qw.endPacketsLost - qw.startPacketsLost
+		if qw.maxRTT > maxRTT {
+			maxRTT = qw.maxRTT
+		}
+		if qw.maxJitter > maxJitter {
+			maxJitter = qw.maxJitter
 		}
-	}
-}
 
-func (cs *ConnectionStats) GetScore() float64 {
-	cs.lock.RLock()
-	defer cs.lock.RUnlock()
-
-	return cs.score
-}
-
-func (cs *ConnectionStats) updateAndGetPercentageLoss() float64 {
-	if cs.params.GetTotalBytes != nil {
-		cs.totalBytes = cs.params.GetTotalBytes()
+		qw.startSeqNum = qw.endSeqNum
+		qw.startPacketsLost = qw.endPacketsLost
+		qw.maxRTT = 0
+		qw.maxJitter = 0
 	}
 
-	if !cs.snapshot.Initialized {
-		cs.snapshot.Initialized = true
-		cs.snapshot.HighestSeqNum = cs.highestSeqNum
-		cs.snapshot.TotalPacketsLost = cs.totalPacketsLost
+	pctLoss := float32(0.0)
+	if int32(lostPacketsInInterval) < 0 {
+		lostPacketsInInterval = 0
 	}
-
-	packetsLostInInterval := cs.totalPacketsLost - cs.snapshot.TotalPacketsLost
-	expectedPacketsInInterval := cs.highestSeqNum - cs.snapshot.HighestSeqNum
-	percentageLoss := float64(0.0)
 	if expectedPacketsInInterval > 0 {
-		percentageLoss = (float64(packetsLostInInterval) / float64(expectedPacketsInInterval)) * 100
+		pctLoss = (float32(lostPacketsInInterval) / float32(expectedPacketsInInterval)) * 100.0
 	}
 
-	cs.snapshot.HighestSeqNum = cs.highestSeqNum
-	cs.snapshot.TotalPacketsLost = cs.totalPacketsLost
+	if cs.params.CodecType == webrtc.RTPCodecTypeAudio {
+		// covert jitter (in media samples units) to milliseconds
+		cs.score = AudioConnectionScore(pctLoss, maxRTT, float32(maxJitter)*1000.0/float32(cs.params.ClockRate))
+	} else {
+		isReducedQuality := false
+		if cs.params.GetIsReducedQuality != nil {
+			isReducedQuality = cs.params.GetIsReducedQuality()
+		}
+		cs.score = VideoConnectionScore(pctLoss, isReducedQuality)
+	}
+}
 
-	return percentageLoss
+func (cs *ConnectionStats) getStat() *livekit.AnalyticsStat {
+	cs.lock.Lock()
+	defer cs.lock.Unlock()
+
+	if cs.params.GetTrackStats == nil {
+		return nil
+	}
+
+	streams := cs.params.GetTrackStats()
+	if len(streams) == 0 {
+		return nil
+	}
+
+	analyticsStreams := make([]*livekit.AnalyticsStream, 0, len(streams))
+	for ssrc, stream := range streams {
+		maxRTT := stream.StreamStats.RTT
+		maxJitter := uint32(stream.StreamStats.Jitter)
+
+		if qw := cs.qualityWindows[ssrc]; qw != nil {
+			maxRTT = qw.maxRTT
+			maxJitter = qw.maxJitter
+		}
+
+		as := ToAnalyticsStream(ssrc, &stream.StreamStats, maxRTT, maxJitter, cs.params.ClockRate)
+
+		//
+		// add video layer if either
+		//   1. Simulcast - even if there is only one layer per stream as it provides layer id
+		//   2. A stream has multiple layers
+		//
+		if cs.params.CodecType == webrtc.RTPCodecTypeVideo && (len(streams) > 1 || len(stream.Layers) > 1) {
+			for layer, layerStats := range stream.Layers {
+				as.VideoLayers = append(as.VideoLayers, ToAnalyticsVideoLayer(layer, &layerStats))
+			}
+		}
+
+		analyticsStreams = append(analyticsStreams, as)
+	}
+
+	cs.updateScore()
+
+	return &livekit.AnalyticsStat{
+		Score:   float32(cs.score),
+		Streams: analyticsStreams,
+	}
 }
 
 func (cs *ConnectionStats) updateStats() {
-	tk := time.NewTicker(cs.params.UpdateInterval)
+	interval := cs.params.UpdateInterval
+	if interval == 0 {
+		interval = connectionQualityUpdateInterval
+	}
+	tk := time.NewTicker(interval)
 	for {
 		select {
 		case <-cs.done:
 			return
 
 		case <-tk.C:
-			cs.lock.Lock()
-			pctLoss := cs.updateAndGetPercentageLoss()
-			if cs.params.CodecType == webrtc.RTPCodecTypeAudio {
-				cs.score = AudioConnectionScore(pctLoss, cs.maxJitter)
-			} else {
-				isReducedQuality := false
-				if cs.params.GetIsReducedQuality != nil {
-					isReducedQuality = cs.params.GetIsReducedQuality()
-				}
-				cs.score = VideoConnectionScore(pctLoss, isReducedQuality)
+			stat := cs.getStat()
+			if stat == nil {
+				continue
 			}
 
-			totalPacketsReceived := uint32(0)
-			if cs.baseSeqNumInitialized {
-				totalPacketsReceived = cs.highestSeqNum - cs.baseSeqNum
-			}
-
-			stat := &livekit.AnalyticsStat{
-				TotalPackets:    uint64(totalPacketsReceived),
-				PacketLost:      uint64(cs.totalPacketsLost),
-				TotalBytes:      cs.totalBytes,
-				Delay:           uint64(cs.maxDelay),
-				Jitter:          float64(cs.maxJitter),
-				NackCount:       cs.nackCount,
-				PliCount:        cs.pliCount,
-				FirCount:        cs.firCount,
-				ConnectionScore: float32(cs.score),
-			}
-			cs.lock.Unlock()
-
 			if cs.onStatsUpdate != nil {
 				cs.onStatsUpdate(cs, stat)
 			}
 		}
 	}
 }
+
+func ToAnalyticsStream(ssrc uint32, streamStats *buffer.StreamStats, maxRTT uint32, maxJitter uint32, clockRate uint32) *livekit.AnalyticsStream {
+	// convert jitter (from number of media samples to microseconds
+	jitter := uint32((float32(maxJitter) * 1e6) / float32(clockRate))
+	return &livekit.AnalyticsStream{
+		Ssrc:                   ssrc,
+		TotalPrimaryPackets:    streamStats.TotalPrimaryPackets,
+		TotalPrimaryBytes:      streamStats.TotalPrimaryBytes,
+		TotalRetransmitPackets: streamStats.TotalRetransmitPackets,
+		TotalRetransmitBytes:   streamStats.TotalRetransmitBytes,
+		TotalPaddingPackets:    streamStats.TotalPaddingPackets,
+		TotalPaddingBytes:      streamStats.TotalPaddingBytes,
+		TotalPacketsLost:       streamStats.TotalPacketsLost,
+		TotalFrames:            streamStats.TotalFrames,
+		Rtt:                    maxRTT,
+		Jitter:                 jitter,
+		TotalNacks:             streamStats.TotalNACKs,
+		TotalPlis:              streamStats.TotalPLIs,
+		TotalFirs:              streamStats.TotalFIRs,
+	}
+}
+
+func ToAnalyticsVideoLayer(layer int, layerStats *buffer.LayerStats) *livekit.AnalyticsVideoLayer {
+	return &livekit.AnalyticsVideoLayer{
+		Layer:        int32(layer),
+		TotalPackets: layerStats.TotalPackets,
+		TotalBytes:   layerStats.TotalBytes,
+		TotalFrames:  layerStats.TotalFrames,
+	}
+}

pkg/sfu/connectionquality/mos.go

@@ -8,10 +8,10 @@ import (
 // available @ https://github.com/oanguenot/webrtc-stats
 
 const (
-	rtt = 70
+	defaultRtt = uint32(70)
 )
 
-func Score2Rating(score float64) livekit.ConnectionQuality {
+func Score2Rating(score float32) livekit.ConnectionQuality {
 	if score > 3.9 {
 		return livekit.ConnectionQuality_EXCELLENT
 	}
@@ -22,12 +22,15 @@ func Score2Rating(score float64) livekit.ConnectionQuality {
 	return livekit.ConnectionQuality_POOR
 }
 
-func mosAudioEmodel(pctLoss float64, jitter uint32) float64 {
+func mosAudioEmodel(pctLoss float32, rtt uint32, jitter float32) float32 {
 	rx := 93.2 - pctLoss
 	ry := 0.18*rx*rx - 27.9*rx + 1126.62
 
-	// Jitter is in MicroSecs (1/1e6) units. Convert it to MilliSecs
-	d := float64(rtt + (jitter / 1000))
+	if rtt == 0 {
+		rtt = defaultRtt
+	}
+	// Jitter is in Milliseconds
+	d := float32(rtt) + jitter
 	h := d - 177.3
 	if h < 0 {
 		h = 0
@@ -47,13 +50,13 @@ func mosAudioEmodel(pctLoss float64, jitter uint32) float64 {
 	return score
 }
 
-func loss2Score(pctLoss float64, reducedQuality bool) float64 {
+func loss2Score(pctLoss float32, reducedQuality bool) float32 {
 	// No Loss, excellent
 	if pctLoss == 0.0 && !reducedQuality {
-		return 5
+		return 5.0
 	}
 	// default when loss is minimal, but reducedQuality
-	score := 3.5
+	score := float32(3.5)
 	// loss is bad
 	if pctLoss >= 4.0 {
 		score = 2.0
@@ -64,10 +67,10 @@ func loss2Score(pctLoss float64, reducedQuality bool) float64 {
 	return score
 }
 
-func AudioConnectionScore(pctLoss float64, jitter uint32) float64 {
-	return mosAudioEmodel(pctLoss, jitter)
+func AudioConnectionScore(pctLoss float32, rtt uint32, jitter float32) float32 {
+	return mosAudioEmodel(pctLoss, rtt, jitter)
 }
 
-func VideoConnectionScore(pctLoss float64, reducedQuality bool) float64 {
+func VideoConnectionScore(pctLoss float32, reducedQuality bool) float32 {
 	return loss2Score(pctLoss, reducedQuality)
 }

pkg/sfu/downtrack.go

@@ -7,7 +7,6 @@ import (
 	"io"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/livekit/protocol/livekit"
@@ -22,10 +21,6 @@ import (
 	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
 )
 
-const (
-	connectionQualityUpdateInterval = 5 * time.Second
-)
-
 // TrackSender defines an interface send media to remote peer
 type TrackSender interface {
 	UpTrackLayersChange(availableLayers []int32)
@@ -66,12 +61,7 @@ var (
 
 type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport)
 
-type PacketStats struct {
-	octets  uint64
-	packets uint32
-}
-
-// DownTrack  implements TrackLocal, is the track used to write packets
+// DownTrack implements TrackLocal, is the track used to write packets
 // to SFU Subscriber, the track handle the packets for simple, simulcast
 // and SVC Publisher.
 type DownTrack struct {
@@ -101,10 +91,8 @@ type DownTrack struct {
 	listenerLock            sync.RWMutex
 	closeOnce               sync.Once
 
-	// Report helpers
-	primaryStats atomic.Value // contains *PacketStats
-	rtxStats     atomic.Value // contains *PacketStats
-	paddingStats atomic.Value // contains *PacketStats
+	statsLock sync.RWMutex
+	stats     buffer.StreamStats
 
 	connectionStats *connectionquality.ConnectionStats
 
@@ -172,12 +160,9 @@ func NewDownTrack(
 	}
 
 	d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{
-		UpdateInterval: connectionQualityUpdateInterval,
-		CodecType:      kind,
-		GetTotalBytes: func() uint64 {
-			octets, _ := d.getSRStats()
-			return octets
-		},
+		CodecType:     kind,
+		ClockRate:     c.ClockRate,
+		GetTrackStats: d.getTrackStats,
 		GetIsReducedQuality: func() bool {
 			return d.GetForwardingStatus() != ForwardingStatusOptimal
 		},
@@ -190,10 +175,6 @@ func NewDownTrack(
 	})
 	d.connectionStats.Start()
 
-	d.primaryStats.Store(new(PacketStats))
-	d.rtxStats.Store(new(PacketStats))
-	d.paddingStats.Store(new(PacketStats))
-
 	return d, nil
 }
 
@@ -285,12 +266,12 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 		}
 	}()
 
-	d.lastRTP.set(time.Now().UnixNano())
-
 	if !d.bound.get() {
 		return nil
 	}
 
+	d.lastRTP.set(time.Now().UnixNano())
+
 	tp, err := d.forwarder.GetTranslationParams(extPkt, layer)
 	if tp.shouldSendPLI {
 		d.lastPli.set(time.Now().UnixNano())
@@ -343,7 +324,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 			d.onMaxLayerChanged(d, layer)
 		}
 
-		d.UpdatePrimaryStats(uint32(pktSize))
+		d.updatePrimaryStats(pktSize, hdr.Marker)
 	} else {
 		d.logger.Errorw("writing rtp packet err", err)
 		d.pktsDropped.add(1)
@@ -355,10 +336,12 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 // WritePaddingRTP tries to write as many padding only RTP packets as necessary
 // to satisfy given size to the DownTrack
 func (d *DownTrack) WritePaddingRTP(bytesToSend int) int {
-	primaryStats := d.primaryStats.Load().(*PacketStats)
-	if primaryStats.packets == 0 {
+	d.statsLock.RLock()
+	if d.stats.TotalPrimaryPackets == 0 {
+		d.statsLock.RUnlock()
 		return 0
 	}
+	d.statsLock.RUnlock()
 
 	// LK-TODO-START
 	// Ideally should look at header extensions negotiated for
@@ -428,7 +411,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int) int {
 		}
 
 		size := hdr.MarshalSize() + len(payload)
-		d.UpdatePaddingStats(uint32(size))
+		d.updatePaddingStats(size)
 		for _, f := range d.onPaddingSent {
 			f(d, size)
 		}
@@ -717,39 +700,41 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport {
 	}
 }
 
-func (d *DownTrack) UpdatePrimaryStats(packetLen uint32) {
-	primaryStats, _ := d.primaryStats.Load().(*PacketStats)
+func (d *DownTrack) updatePrimaryStats(packetLen int, marker bool) {
+	d.statsLock.Lock()
+	defer d.statsLock.Unlock()
 
-	primaryStats.octets += uint64(packetLen)
-	primaryStats.packets += 1
-
-	d.primaryStats.Store(primaryStats)
+	d.stats.TotalPrimaryPackets++
+	d.stats.TotalPrimaryBytes += uint64(packetLen)
+	if marker {
+		d.stats.TotalFrames++
+	}
 }
 
-func (d *DownTrack) UpdateRtxStats(packetLen uint32) {
-	rtxStats, _ := d.rtxStats.Load().(*PacketStats)
+func (d *DownTrack) updateRtxStats(packetLen int) {
+	d.statsLock.Lock()
+	defer d.statsLock.Unlock()
 
-	rtxStats.octets += uint64(packetLen)
-	rtxStats.packets += 1
-
-	d.rtxStats.Store(rtxStats)
+	d.stats.TotalRetransmitPackets++
+	d.stats.TotalRetransmitBytes += uint64(packetLen)
 }
 
-func (d *DownTrack) UpdatePaddingStats(packetLen uint32) {
-	paddingStats, _ := d.paddingStats.Load().(*PacketStats)
+func (d *DownTrack) updatePaddingStats(packetLen int) {
+	d.statsLock.Lock()
+	defer d.statsLock.Unlock()
 
-	paddingStats.octets += uint64(packetLen)
-	paddingStats.packets += 1
-
-	d.paddingStats.Store(paddingStats)
+	d.stats.TotalPaddingPackets++
+	d.stats.TotalPaddingBytes += uint64(packetLen)
 }
 
 func (d *DownTrack) writeBlankFrameRTP() error {
 	// don't send if nothing has been sent
-	primaryStats := d.primaryStats.Load().(*PacketStats)
-	if primaryStats.packets == 0 {
+	d.statsLock.RLock()
+	if d.stats.TotalPrimaryPackets == 0 {
+		d.statsLock.RUnlock()
 		return nil
 	}
+	d.statsLock.RUnlock()
 
 	// LK-TODO: Support other video codecs
 	if d.kind == webrtc.RTPCodecTypeAudio || (d.mime != "video/vp8" && d.mime != "video/h264") {
@@ -798,7 +783,7 @@ func (d *DownTrack) writeBlankFrameRTP() error {
 			f(d, pktSize)
 		}
 
-		d.UpdatePrimaryStats(uint32(pktSize))
+		d.updatePrimaryStats(pktSize, hdr.Marker)
 
 		// only the first frame will need frameEndNeeded to close out the
 		// previous picture, rest are small key frames
@@ -865,12 +850,17 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 		}
 	}
 
+	var numNACKs uint32
+	var numPLIs uint32
+	var numFIRs uint32
 	for _, pkt := range pkts {
 		switch p := pkt.(type) {
 		case *rtcp.PictureLossIndication:
+			numPLIs++
 			sendPliOnce()
 
 		case *rtcp.FullIntraRequest:
+			numFIRs++
 			sendPliOnce()
 
 		case *rtcp.ReceiverEstimatedMaximumBitrate:
@@ -889,6 +879,14 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 					continue
 				}
 				rr.Reports = append(rr.Reports, r)
+
+				d.statsLock.Lock()
+				d.stats.TotalPacketsLost = r.TotalLost
+				// RAJA-TODO - calculate RTT and update
+				d.stats.Jitter = float64(r.Jitter)
+
+				d.connectionStats.UpdateWindow(r.SSRC, r.LastSequenceNumber, r.TotalLost, 0, r.Jitter)
+				d.statsLock.Unlock()
 			}
 			if len(rr.Reports) > 0 {
 				d.listenerLock.RLock()
@@ -904,6 +902,7 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 				nackedPackets = append(nackedPackets, d.sequencer.getSeqNoPairs(pair.PacketList())...)
 			}
 			go d.retransmitPackets(nackedPackets)
+			numNACKs += uint32(len(nackedPackets))
 
 		case *rtcp.TransportLayerCC:
 			if p.MediaSSRC == d.ssrc && d.onTransportCCFeedback != nil {
@@ -912,7 +911,11 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 		}
 	}
 
-	d.connectionStats.RTCPFeedback(pkts, d.ssrc)
+	d.statsLock.Lock()
+	d.stats.TotalNACKs += numNACKs
+	d.stats.TotalPLIs += numPLIs
+	d.stats.TotalFIRs += numFIRs
+	d.statsLock.Unlock()
 }
 
 func (d *DownTrack) retransmitPackets(nackedPackets []packetMeta) {
@@ -985,17 +988,19 @@ func (d *DownTrack) retransmitPackets(nackedPackets []packetMeta) {
 				f(d, pktSize)
 			}
 
-			d.UpdateRtxStats(uint32(pktSize))
+			d.updateRtxStats(pktSize)
 		}
 	}
 }
 
 func (d *DownTrack) getSRStats() (uint64, uint32) {
-	primary := d.primaryStats.Load().(*PacketStats)
-	rtx := d.rtxStats.Load().(*PacketStats)
-	padding := d.paddingStats.Load().(*PacketStats)
+	d.statsLock.RLock()
+	defer d.statsLock.RUnlock()
 
-	return primary.octets + rtx.octets + padding.octets, primary.packets + rtx.packets + padding.packets
+	packets := d.stats.TotalPrimaryPackets + d.stats.TotalRetransmitPackets + d.stats.TotalPaddingPackets
+	octets := d.stats.TotalPrimaryBytes + d.stats.TotalRetransmitBytes + d.stats.TotalPaddingBytes
+
+	return octets, packets
 }
 
 // writes RTP header extensions of track
@@ -1091,6 +1096,27 @@ func (d *DownTrack) DebugInfo() map[string]interface{} {
 	}
 }
 
-func (d *DownTrack) GetConnectionScore() float64 {
+func (d *DownTrack) GetConnectionScore() float32 {
 	return d.connectionStats.GetScore()
 }
+
+func (d *DownTrack) getTrackStats() map[uint32]*buffer.StreamStatsWithLayers {
+	d.statsLock.RLock()
+	defer d.statsLock.RUnlock()
+
+	stats := make(map[uint32]*buffer.StreamStatsWithLayers, 1)
+
+	layers := make(map[int]buffer.LayerStats)
+	layers[0] = buffer.LayerStats{
+		TotalPackets: d.stats.TotalPrimaryPackets + d.stats.TotalRetransmitPackets + d.stats.TotalPaddingPackets,
+		TotalBytes:   d.stats.TotalPrimaryBytes + d.stats.TotalRetransmitBytes + d.stats.TotalPaddingBytes,
+		TotalFrames:  d.stats.TotalFrames,
+	}
+
+	stats[d.ssrc] = &buffer.StreamStatsWithLayers{
+		StreamStats: d.stats,
+		Layers:      layers,
+	}
+
+	return stats
+}

pkg/sfu/receiver.go

@@ -2,7 +2,6 @@ package sfu
 
 import (
 	"io"
-	"math/rand"
 	"runtime"
 	"sync"
 	"sync/atomic"
@@ -15,6 +14,7 @@ import (
 	"github.com/rs/zerolog/log"
 
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
 )
 
 type AudioLevelHandle func(level uint8, duration uint32)
@@ -76,6 +76,12 @@ type WebRTCReceiver struct {
 	lbThreshold int
 
 	streamTrackerManager *StreamTrackerManager
+
+	rtt             uint32
+	connectionStats *connectionquality.ConnectionStats
+
+	// update stats
+	onStatsUpdate func(w *WebRTCReceiver, stat *livekit.AnalyticsStat)
 }
 
 func RidToLayer(rid string) int32 {
@@ -145,15 +151,52 @@ func NewWebRTCReceiver(
 		streamTrackerManager: NewStreamTrackerManager(),
 	}
 	w.streamTrackerManager.OnAvailableLayersChanged(w.downTrackLayerChange)
+
 	if runtime.GOMAXPROCS(0) < w.numProcs {
 		w.numProcs = runtime.GOMAXPROCS(0)
 	}
+
 	for _, opt := range opts {
 		w = opt(w)
 	}
+
+	w.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{
+		CodecType:     w.kind,
+		ClockRate:     w.codec.ClockRate,
+		GetTrackStats: w.getTrackStats,
+		GetIsReducedQuality: func() bool {
+			return w.streamTrackerManager.IsReducedQuality()
+		},
+		Logger: w.logger,
+	})
+	w.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) {
+		if w.onStatsUpdate != nil {
+			w.onStatsUpdate(w, stat)
+		}
+	})
+	w.connectionStats.Start()
+
 	return w
 }
 
+func (w *WebRTCReceiver) OnStatsUpdate(fn func(w *WebRTCReceiver, stat *livekit.AnalyticsStat)) {
+	w.onStatsUpdate = fn
+}
+
+func (w *WebRTCReceiver) GetConnectionScore() float32 {
+	return w.connectionStats.GetScore()
+}
+
+func (w *WebRTCReceiver) SetRTT(rtt uint32) {
+	w.rtt = rtt
+
+	w.bufferMu.RLock()
+	for _, buffer := range w.buffers {
+		buffer.SetRTT(rtt)
+	}
+	w.bufferMu.RUnlock()
+}
+
 func (w *WebRTCReceiver) SetTrackMeta(trackID livekit.TrackID, streamID string) {
 	w.streamID = streamID
 	w.trackID = trackID
@@ -191,6 +234,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff
 	}
 
 	buff.SetLogger(w.logger)
+	buff.OnFeedback(w.sendRTCP)
 
 	layer := RidToLayer(track.RID())
 
@@ -242,10 +286,6 @@ func (w *WebRTCReceiver) SetMaxExpectedSpatialLayer(layer int32) {
 	w.streamTrackerManager.SetMaxExpectedSpatialLayer(layer)
 }
 
-func (w *WebRTCReceiver) NumAvailableSpatialLayers() int {
-	return len(w.streamTrackerManager.GetAvailableLayers())
-}
-
 func (w *WebRTCReceiver) downTrackLayerChange(layers []int32) {
 	w.downTrackMu.RLock()
 	downTracks := w.downTracks
@@ -301,26 +341,34 @@ func (w *WebRTCReceiver) DeleteDownTrack(peerID livekit.ParticipantID) {
 	w.free[idx] = struct{}{}
 }
 
-func (w *WebRTCReceiver) SendRTCP(p []rtcp.Packet) {
-	if _, ok := p[0].(*rtcp.PictureLossIndication); ok {
-		w.rtcpMu.Lock()
-		throttled := time.Now().UnixNano()-w.lastPli.get() < w.pliThrottle
-		w.rtcpMu.Unlock()
-		if throttled {
-			return
-		}
-		w.lastPli.set(time.Now().UnixNano())
-	}
+func (w *WebRTCReceiver) sendRTCP(packets []rtcp.Packet) {
+	w.rtcpCh <- packets
 
-	w.rtcpCh <- p
+	for _, p := range packets {
+		switch pkt := p.(type) {
+		case *rtcp.ReceiverReport:
+			for _, r := range pkt.Reports {
+				w.connectionStats.UpdateWindow(r.SSRC, r.LastSequenceNumber, r.TotalLost, w.rtt, r.Jitter)
+			}
+		}
+	}
 }
 
 func (w *WebRTCReceiver) SendPLI(layer int32) {
-	pli := []rtcp.Packet{
-		&rtcp.PictureLossIndication{SenderSSRC: rand.Uint32(), MediaSSRC: w.SSRC(int(layer))},
+	w.bufferMu.RLock()
+	buff := w.buffers[layer]
+	w.bufferMu.RUnlock()
+	if buff == nil {
+		return
 	}
 
-	w.SendRTCP(pli)
+	throttled := time.Now().UnixNano()-w.lastPli.get() < w.pliThrottle
+	if throttled {
+		return
+	}
+	w.lastPli.set(time.Now().UnixNano())
+
+	buff.SendPLI()
 }
 
 func (w *WebRTCReceiver) SetRTCPCh(ch chan []rtcp.Packet) {
@@ -343,19 +391,30 @@ func (w *WebRTCReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error
 	return buff.GetPacket(buf, sn)
 }
 
-func (w *WebRTCReceiver) GetTotalBytes() uint64 {
+func (w *WebRTCReceiver) getTrackStats() map[uint32]*buffer.StreamStatsWithLayers {
 	w.bufferMu.RLock()
 	defer w.bufferMu.RUnlock()
 
-	totalBytes := uint64(0)
-	for _, buff := range w.buffers {
-		if buff != nil {
-			stats := buff.GetStats()
-			totalBytes += stats.TotalBytes
+	stats := make(map[uint32]*buffer.StreamStatsWithLayers, len(w.buffers))
+
+	for layer, buff := range w.buffers {
+		if buff == nil {
+			continue
 		}
+
+		sswl := buff.GetStats()
+
+		// if simulcast, patch buffer stats with correct layer
+		if w.isSimulcast {
+			patched := make(map[int]buffer.LayerStats, 1)
+			patched[layer] = sswl.Layers[0]
+			sswl.Layers = patched
+		}
+
+		stats[w.SSRC(layer)] = sswl
 	}
 
-	return totalBytes
+	return stats
 }
 
 func (w *WebRTCReceiver) forwardRTP(layer int32) {
@@ -435,6 +494,8 @@ func (w *WebRTCReceiver) writeRTP(layer int32, dt TrackSender, pkt *buffer.ExtPa
 
 // closeTracks close all tracks from Receiver
 func (w *WebRTCReceiver) closeTracks() {
+	w.connectionStats.Close()
+
 	w.downTrackMu.Lock()
 	for _, dt := range w.downTracks {
 		if dt != nil {

pkg/sfu/streamtrackermanager.go

@@ -134,6 +134,13 @@ func (s *StreamTrackerManager) SetMaxExpectedSpatialLayer(layer int32) {
 	}
 }
 
+func (s *StreamTrackerManager) IsReducedQuality() bool {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	return int32(len(s.availableLayers)) < (s.maxExpectedLayer + 1)
+}
+
 func (s *StreamTrackerManager) GetAvailableLayers() []int32 {
 	s.lock.RLock()
 	defer s.lock.RUnlock()

pkg/telemetry/prometheus/packets.go

@@ -86,7 +86,7 @@ func IncrementBytes(direction Direction, count uint64) {
 	}
 }
 
-func IncrementRTCP(direction Direction, nack, pli, fir int32) {
+func IncrementRTCP(direction Direction, nack, pli, fir uint32) {
 	if nack > 0 {
 		promNackTotal.WithLabelValues(string(direction)).Add(float64(nack))
 		atomic.AddUint64(&atomicNackTotal, uint64(nack))

pkg/telemetry/statsworker.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/livekit/protocol/livekit"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -15,23 +16,17 @@ type StatsWorker struct {
 	roomName      livekit.RoomName
 	participantID livekit.ParticipantID
 
-	drainStats       map[livekit.TrackID]bool
-	outgoingPerTrack map[livekit.TrackID]*Stats
-	incomingPerTrack map[livekit.TrackID]*Stats
+	outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
+	incomingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
 }
 
-type Stats struct {
-	next             *livekit.AnalyticsStat
-	totalPackets     uint32
-	prevPackets      uint32
-	totalBytes       uint64
-	prevBytes        uint64
-	totalPacketsLost uint64
-	prevPacketsLost  uint64
-	connectionScore  float32
-}
-
-func newStatsWorker(ctx context.Context, t TelemetryReporter, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID) *StatsWorker {
+func newStatsWorker(
+	ctx context.Context,
+	t TelemetryReporter,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
+	participantID livekit.ParticipantID,
+) *StatsWorker {
 	s := &StatsWorker{
 		ctx:           ctx,
 		t:             t,
@@ -39,134 +34,146 @@ func newStatsWorker(ctx context.Context, t TelemetryReporter, roomID livekit.Roo
 		roomName:      roomName,
 		participantID: participantID,
 
-		outgoingPerTrack: make(map[livekit.TrackID]*Stats),
-		incomingPerTrack: make(map[livekit.TrackID]*Stats),
-		drainStats:       make(map[livekit.TrackID]bool),
+		outgoingPerTrack: make(map[livekit.TrackID][]*livekit.AnalyticsStat),
+		incomingPerTrack: make(map[livekit.TrackID][]*livekit.AnalyticsStat),
 	}
 	return s
 }
 
-func (s *StatsWorker) getOrCreateOutgoingStatsIfEmpty(trackID livekit.TrackID) *Stats {
-	if s.outgoingPerTrack[trackID] == nil {
-		s.outgoingPerTrack[trackID] = &Stats{next: &livekit.AnalyticsStat{
-			Kind:          livekit.StreamType_DOWNSTREAM,
-			RoomId:        string(s.roomID),
-			ParticipantId: string(s.participantID),
-			RoomName:      string(s.roomName),
-		}}
-	}
-	return s.outgoingPerTrack[trackID]
+func (s *StatsWorker) appendOutgoing(trackID livekit.TrackID, stat *livekit.AnalyticsStat) {
+	s.outgoingPerTrack[trackID] = append(s.outgoingPerTrack[trackID], stat)
 }
 
-func (s *StatsWorker) getOrCreateIncomingStatsIfEmpty(trackID livekit.TrackID) *Stats {
-	if s.incomingPerTrack[trackID] == nil {
-		s.incomingPerTrack[trackID] = &Stats{next: &livekit.AnalyticsStat{
-			Kind:          livekit.StreamType_UPSTREAM,
-			RoomId:        string(s.roomID),
-			ParticipantId: string(s.participantID),
-			RoomName:      string(s.roomName),
-		}}
-	}
-	return s.incomingPerTrack[trackID]
+func (s *StatsWorker) appendIncoming(trackID livekit.TrackID, stat *livekit.AnalyticsStat) {
+	s.incomingPerTrack[trackID] = append(s.incomingPerTrack[trackID], stat)
 }
 
-func (s *StatsWorker) OnTrackStat(trackID livekit.TrackID, direction livekit.StreamType, stats *livekit.AnalyticsStat) {
-	var ds *Stats
+func (s *StatsWorker) OnTrackStat(trackID livekit.TrackID, direction livekit.StreamType, stat *livekit.AnalyticsStat) {
 	if direction == livekit.StreamType_DOWNSTREAM {
-		ds = s.getOrCreateOutgoingStatsIfEmpty(trackID)
+		s.appendOutgoing(trackID, stat)
 	} else {
-		ds = s.getOrCreateIncomingStatsIfEmpty(trackID)
-	}
-	ds.totalPacketsLost = stats.PacketLost
-	ds.totalPackets = uint32(stats.TotalPackets)
-	ds.totalBytes = stats.TotalBytes
-
-	if stats.Rtt > ds.next.Rtt {
-		ds.next.Rtt = stats.Rtt
-	}
-	if stats.Jitter > ds.next.Jitter {
-		ds.next.Jitter = stats.Jitter
-	}
-	ds.next.NackCount += stats.NackCount
-	ds.next.PliCount += stats.PliCount
-	ds.next.FirCount += stats.FirCount
-	// average out scores received in this interval
-	if ds.connectionScore == 0 {
-		ds.connectionScore = stats.ConnectionScore
-	} else {
-		ds.connectionScore = (ds.connectionScore + stats.ConnectionScore) / 2
+		s.appendIncoming(trackID, stat)
 	}
 }
 
 func (s *StatsWorker) Update() {
 	ts := timestamppb.Now()
-	stats := make([]*livekit.AnalyticsStat, 0)
 
+	stats := make([]*livekit.AnalyticsStat, 0, len(s.incomingPerTrack)+len(s.outgoingPerTrack))
 	stats = s.collectUpstreamStats(ts, stats)
 	stats = s.collectDownstreamStats(ts, stats)
-
-	s.t.Report(s.ctx, stats)
+	if len(stats) > 0 {
+		s.t.Report(s.ctx, stats)
+	}
 }
 
 func (s *StatsWorker) collectDownstreamStats(ts *timestamppb.Timestamp, stats []*livekit.AnalyticsStat) []*livekit.AnalyticsStat {
-	for trackID, trackDownStreamStats := range s.outgoingPerTrack {
-		analyticsStat := s.update(trackDownStreamStats, ts)
-		if analyticsStat != nil {
-			analyticsStat.TrackId = string(trackID)
-			stats = append(stats, analyticsStat)
+	for trackID, analyticsStats := range s.outgoingPerTrack {
+		analyticsStat := coalesce(analyticsStats)
+		if analyticsStat == nil {
+			continue
 		}
+
+		s.patch(analyticsStat, ts, trackID, livekit.StreamType_DOWNSTREAM)
+		stats = append(stats, analyticsStat)
 	}
-	if len(s.drainStats) > 0 {
-		for trackID := range s.drainStats {
-			delete(s.outgoingPerTrack, trackID)
-			delete(s.incomingPerTrack, trackID)
-		}
-		s.drainStats = make(map[livekit.TrackID]bool)
-	}
+	s.outgoingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat, 0)
 
 	return stats
 }
 
 func (s *StatsWorker) collectUpstreamStats(ts *timestamppb.Timestamp, stats []*livekit.AnalyticsStat) []*livekit.AnalyticsStat {
-	for trackID, trackUpStreamStats := range s.incomingPerTrack {
-		analyticsStat := s.update(trackUpStreamStats, ts)
-		if analyticsStat != nil {
-			analyticsStat.TrackId = string(trackID)
-			stats = append(stats, analyticsStat)
+	for trackID, analyticsStats := range s.incomingPerTrack {
+		analyticsStat := coalesce(analyticsStats)
+		if analyticsStat == nil {
+			continue
 		}
+
+		s.patch(analyticsStat, ts, trackID, livekit.StreamType_UPSTREAM)
+		stats = append(stats, analyticsStat)
 	}
+	s.incomingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat, 0)
+
 	return stats
 }
 
-func (s *StatsWorker) update(stats *Stats, ts *timestamppb.Timestamp) *livekit.AnalyticsStat {
-	if stats.totalBytes == 0 {
-		return nil
-	}
-
-	next := stats.next
-	stats.next = &livekit.AnalyticsStat{
-		Kind:          next.Kind,
-		RoomId:        string(s.roomID),
-		ParticipantId: string(s.participantID),
-		RoomName:      string(s.roomName),
-	}
-
-	next.TimeStamp = ts
-	next.TotalPackets = uint64(stats.totalPackets - stats.prevPackets)
-	next.TotalBytes = stats.totalBytes - stats.prevBytes
-	next.PacketLost = stats.totalPacketsLost - stats.prevPacketsLost
-	next.ConnectionScore = stats.connectionScore
-
-	stats.prevPackets = stats.totalPackets
-	stats.prevBytes = stats.totalBytes
-	stats.prevPacketsLost = stats.totalPacketsLost
-	return next
+func (s *StatsWorker) patch(
+	analyticsStat *livekit.AnalyticsStat,
+	ts *timestamppb.Timestamp,
+	trackID livekit.TrackID,
+	kind livekit.StreamType,
+) {
+	analyticsStat.TimeStamp = ts
+	analyticsStat.TrackId = string(trackID)
+	analyticsStat.Kind = kind
+	analyticsStat.RoomId = string(s.roomID)
+	analyticsStat.ParticipantId = string(s.participantID)
+	analyticsStat.RoomName = string(s.roomName)
 }
 
 func (s *StatsWorker) Close() {
 	s.Update()
 }
 
-func (s *StatsWorker) RemoveStats(trackID livekit.TrackID) {
-	s.drainStats[trackID] = true
+func coalesce(stats []*livekit.AnalyticsStat) *livekit.AnalyticsStat {
+	if len(stats) == 0 {
+		return nil
+	}
+
+	// average score of all available stats
+	score := float32(0.0)
+	for _, stat := range stats {
+		score += stat.Score
+	}
+	score = score / float32(len(stats))
+
+	// aggregate streams across all stats
+	maxRTT := make(map[uint32]uint32)
+	maxJitter := make(map[uint32]uint32)
+	analyticsStreams := make(map[uint32]*livekit.AnalyticsStream)
+	for _, stat := range stats {
+		//
+		// For each stream (identified by SSRC) consolidate reports.
+		// For cumulative stats, take the latest report.
+		// For instantaneous stats, take maximum (or some other appropriate representation)
+		//
+		for _, stream := range stat.Streams {
+			ssrc := stream.Ssrc
+			analyticsStream := analyticsStreams[ssrc]
+			if analyticsStream == nil {
+				analyticsStreams[ssrc] = stream
+				maxRTT[ssrc] = stream.Rtt
+				maxJitter[ssrc] = stream.Jitter
+				continue
+			}
+
+			if stream.TotalPrimaryPackets <= analyticsStream.TotalPrimaryPackets {
+				// total count should be monotonically increasing
+				continue
+			}
+
+			analyticsStreams[ssrc] = stream
+			if stream.Rtt > maxRTT[ssrc] {
+				maxRTT[ssrc] = stream.Rtt
+			}
+
+			if stream.Jitter > maxJitter[ssrc] {
+				maxJitter[ssrc] = stream.Jitter
+			}
+		}
+	}
+
+	streams := make([]*livekit.AnalyticsStream, 0, len(analyticsStreams))
+	for ssrc, analyticsStream := range analyticsStreams {
+		stream := proto.Clone(analyticsStream).(*livekit.AnalyticsStream)
+		stream.Rtt = maxRTT[ssrc]
+		stream.Jitter = maxJitter[ssrc]
+
+		streams = append(streams, stream)
+	}
+
+	return &livekit.AnalyticsStat{
+		Score:   score,
+		Streams: streams,
+	}
 }

pkg/telemetry/telemetryserviceinternal.go

@@ -4,10 +4,9 @@ import (
 	"context"
 
 	"github.com/gammazero/workerpool"
+	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/webhook"
-
-	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 )
 
 type TelemetryServiceInternal interface {
@@ -45,7 +44,15 @@ func (t *telemetryServiceInternal) TrackStats(streamType livekit.StreamType, par
 		direction = prometheus.Outgoing
 	}
 
-	prometheus.IncrementRTCP(direction, stat.NackCount, stat.PliCount, stat.FirCount)
+	totalNACKs := uint32(0)
+	totalPLIs := uint32(0)
+	totalFIRs := uint32(0)
+	for _, stream := range stat.Streams {
+		totalNACKs += stream.TotalNacks
+		totalPLIs += stream.TotalPlis
+		totalFIRs += stream.TotalFirs
+	}
+	prometheus.IncrementRTCP(direction, totalNACKs, totalPLIs, totalFIRs)
 
 	w := t.workers[participantID]
 	if w != nil {
@@ -55,13 +62,23 @@ func (t *telemetryServiceInternal) TrackStats(streamType livekit.StreamType, par
 
 func (t *telemetryServiceInternal) Report(ctx context.Context, stats []*livekit.AnalyticsStat) {
 	for _, stat := range stats {
+		if len(stat.Streams) == 0 {
+			continue
+		}
+
 		direction := prometheus.Incoming
 		if stat.Kind == livekit.StreamType_DOWNSTREAM {
 			direction = prometheus.Outgoing
 		}
 
-		prometheus.IncrementPackets(direction, stat.TotalPackets)
-		prometheus.IncrementBytes(direction, stat.TotalBytes)
+		totalPackets := uint32(0)
+		totalBytes := uint64(0)
+		for _, stream := range stat.Streams {
+			totalPackets += (stream.TotalPrimaryPackets + stream.TotalRetransmitPackets + stream.TotalPaddingPackets)
+			totalBytes += (stream.TotalPrimaryBytes + stream.TotalRetransmitBytes + stream.TotalPaddingBytes)
+		}
+		prometheus.IncrementPackets(direction, uint64(totalPackets))
+		prometheus.IncrementBytes(direction, totalBytes)
 	}
 
 	t.analytics.SendStats(ctx, stats)

pkg/telemetry/telemetryserviceinternalevents.go

@@ -126,7 +126,6 @@ func (t *telemetryServiceInternal) TrackUnpublished(ctx context.Context, partici
 	if w != nil {
 		roomID = w.roomID
 		roomName = w.roomName
-		w.RemoveStats(livekit.TrackID(track.GetSid()))
 	}
 
 	prometheus.SubPublishedTrack(track.Type.String())

pkg/telemetry/test/telemetry_service_test.go

@@ -35,7 +35,7 @@ func Test_ParticipantAndRoomDataAreSentWithAnalytics(t *testing.T) {
 
 	// do
 	packet := 33
-	stat := &livekit.AnalyticsStat{TotalBytes: uint64(packet)}
+	stat := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{TotalPrimaryBytes: uint64(packet)}}}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, "", stat)
 	fixture.sut.SendAnalytics()
 
@@ -67,7 +67,7 @@ func Test_OnDownstreamPackets(t *testing.T) {
 	var bytes int
 	for i := range packets {
 		bytes += packets[i]
-		stat := &livekit.AnalyticsStat{TotalBytes: uint64(bytes), TotalPackets: uint64(i + 1)}
+		stat := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{TotalPrimaryBytes: uint64(bytes), TotalPrimaryPackets: uint32(i + 1)}}}
 		fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat)
 	}
 	fixture.sut.SendAnalytics()
@@ -77,8 +77,8 @@ func Test_OnDownstreamPackets(t *testing.T) {
 	_, stats := fixture.analytics.SendStatsArgsForCall(0)
 	require.Equal(t, 1, len(stats))
 	require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind)
-	require.Equal(t, totalBytes, int(stats[0].TotalBytes))
-	require.Equal(t, totalPackets, int(stats[0].TotalPackets))
+	require.Equal(t, totalBytes, int(stats[0].Streams[0].TotalPrimaryBytes))
+	require.Equal(t, totalPackets, int(stats[0].Streams[0].TotalPrimaryPackets))
 	require.Equal(t, string(trackID), stats[0].TrackId)
 }
 
@@ -95,12 +95,12 @@ func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) {
 	// do
 	packet1 := 33
 	trackID1 := livekit.TrackID("trackID1")
-	stat1 := &livekit.AnalyticsStat{TotalBytes: uint64(packet1), TotalPackets: 1}
+	stat1 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{TotalPrimaryBytes: uint64(packet1), TotalPrimaryPackets: 1}}}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID1, stat1)
 
 	packet2 := 23
 	trackID2 := livekit.TrackID("trackID2")
-	stat2 := &livekit.AnalyticsStat{TotalBytes: uint64(packet2), TotalPackets: 1}
+	stat2 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{TotalPrimaryBytes: uint64(packet2), TotalPrimaryPackets: 1}}}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID2, stat2)
 	fixture.sut.SendAnalytics()
 
@@ -114,19 +114,19 @@ func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) {
 	for _, sentStat := range stats {
 		if livekit.TrackID(sentStat.TrackId) == trackID1 {
 			found1 = true
-			require.Equal(t, packet1, int(sentStat.TotalBytes))
-			require.Equal(t, 1, int(sentStat.TotalPackets))
+			require.Equal(t, packet1, int(sentStat.Streams[0].TotalPrimaryBytes))
+			require.Equal(t, 1, int(sentStat.Streams[0].TotalPrimaryPackets))
 		} else if livekit.TrackID(sentStat.TrackId) == trackID2 {
 			found2 = true
-			require.Equal(t, packet2, int(sentStat.TotalBytes))
-			require.Equal(t, 1, int(sentStat.TotalPackets))
+			require.Equal(t, packet2, int(sentStat.Streams[0].TotalPrimaryBytes))
+			require.Equal(t, 1, int(sentStat.Streams[0].TotalPrimaryPackets))
 		}
 	}
 	require.True(t, found1)
 	require.True(t, found2)
 }
 
-func Test_OnDownStreamRTCP(t *testing.T) {
+func Test_OnDownStreamStat(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
@@ -136,10 +136,34 @@ func Test_OnDownStreamRTCP(t *testing.T) {
 	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil)
 
 	// do
-	stat1 := &livekit.AnalyticsStat{NackCount: 1, PliCount: 1, Jitter: 3, PacketLost: 3, TotalBytes: 1, TotalPackets: 1}
+	stat1 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   1,
+				TotalPrimaryPackets: 1,
+				TotalPacketsLost:    3,
+				TotalNacks:          1,
+				TotalPlis:           1,
+				Jitter:              3,
+			},
+		},
+	}
 	trackID := livekit.TrackID("trackID1")
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat1)
-	stat2 := &livekit.AnalyticsStat{FirCount: 1, Jitter: 5, PacketLost: 4, TotalBytes: 2, TotalPackets: 2}
+
+	stat2 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   2,
+				TotalPrimaryPackets: 2,
+				TotalPacketsLost:    4,
+				TotalNacks:          1,
+				TotalPlis:           1,
+				TotalFirs:           1,
+				Jitter:              5,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat2)
 
 	fixture.sut.SendAnalytics()
@@ -149,12 +173,12 @@ func Test_OnDownStreamRTCP(t *testing.T) {
 	_, stats := fixture.analytics.SendStatsArgsForCall(0)
 	require.Equal(t, 1, len(stats))
 	require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind)
-	require.Equal(t, 1, int(stats[0].NackCount))
-	require.Equal(t, 1, int(stats[0].PliCount))
-	require.Equal(t, 1, int(stats[0].FirCount))
-	require.Equal(t, 0, int(stats[0].Rtt))        // TODO: test for RTT
-	require.Equal(t, 5, int(stats[0].Jitter))     // max of jitter, see list of rtcp.ReceptionReport above
-	require.Equal(t, 4, int(stats[0].PacketLost)) // last reported packets lost, see list of rtcp.ReceptionReport above
+	require.Equal(t, 1, int(stats[0].Streams[0].TotalNacks))
+	require.Equal(t, 1, int(stats[0].Streams[0].TotalPlis))
+	require.Equal(t, 1, int(stats[0].Streams[0].TotalFirs))
+	require.Equal(t, 0, int(stats[0].Streams[0].Rtt))              // TODO: test for RTT
+	require.Equal(t, 5, int(stats[0].Streams[0].Jitter))           // max of jitter, see list of rtcp.ReceptionReport above
+	require.Equal(t, 4, int(stats[0].Streams[0].TotalPacketsLost)) // last reported packets lost, see list of rtcp.ReceptionReport above
 	require.Equal(t, string(trackID), stats[0].TrackId)
 }
 
@@ -169,10 +193,26 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) {
 
 	// do
 	trackID := livekit.TrackID("trackID1")
-	stat1 := &livekit.AnalyticsStat{PacketLost: 1, TotalPackets: 1, TotalBytes: 1}
+	stat1 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   1,
+				TotalPrimaryPackets: 1,
+				TotalPacketsLost:    1,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat1) // there should be bytes reported so that stats are sent
 	fixture.sut.SendAnalytics()
-	stat2 := &livekit.AnalyticsStat{PacketLost: 4, TotalPackets: 2, TotalBytes: 2}
+	stat2 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   2,
+				TotalPrimaryPackets: 2,
+				TotalPacketsLost:    4,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat2)
 	fixture.sut.SendAnalytics()
 
@@ -181,12 +221,12 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) {
 	_, stats := fixture.analytics.SendStatsArgsForCall(0)
 	require.Equal(t, 1, len(stats))
 	require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind)
-	require.Equal(t, 1, int(stats[0].PacketLost)) // see pkts1
+	require.Equal(t, 1, int(stats[0].Streams[0].TotalPacketsLost)) // see pkts1
 
 	_, stats = fixture.analytics.SendStatsArgsForCall(1)
 	require.Equal(t, 1, len(stats))
 	require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind)
-	require.Equal(t, 3, int(stats[0].PacketLost)) // see diff of TotalLost between pkts2 and pkts1
+	require.Equal(t, 4, int(stats[0].Streams[0].TotalPacketsLost)) // see diff of TotalLost between pkts2 and pkts1
 }
 
 func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
@@ -202,13 +242,36 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
 
 	trackID1 := livekit.TrackID("trackID1")
 	trackID2 := livekit.TrackID("trackID2")
-	stat1 := &livekit.AnalyticsStat{TotalBytes: 1, TotalPackets: 1}
+	stat1 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   1,
+				TotalPrimaryPackets: 1,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID1, stat1) // there should be bytes reported so that stats are sent
 
-	stat2 := &livekit.AnalyticsStat{NackCount: 1, TotalPackets: 2, TotalBytes: 2}
+	stat2 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   2,
+				TotalPrimaryPackets: 2,
+				TotalNacks:          1,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID1, stat2)
 
-	stat3 := &livekit.AnalyticsStat{FirCount: 1, TotalPackets: 3, TotalBytes: 3}
+	stat3 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   3,
+				TotalPrimaryPackets: 3,
+				TotalFirs:           1,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID2, stat3)
 	fixture.sut.SendAnalytics()
 
@@ -223,17 +286,17 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
 		if livekit.TrackID(sentStat.TrackId) == trackID1 {
 			found1 = true
 			require.Equal(t, livekit.StreamType_DOWNSTREAM, sentStat.Kind)
-			require.Equal(t, 1, int(sentStat.NackCount)) // see pkts1 above
+			require.Equal(t, 1, int(sentStat.Streams[0].TotalNacks)) // see pkts1 above
 		} else if livekit.TrackID(sentStat.TrackId) == trackID2 {
 			found2 = true
 			require.Equal(t, livekit.StreamType_DOWNSTREAM, sentStat.Kind)
-			require.Equal(t, 1, int(sentStat.FirCount)) // see pkts2 above
+			require.Equal(t, 1, int(sentStat.Streams[0].TotalFirs)) // see pkts2 above
 		}
 	}
 	require.True(t, found1)
 	require.True(t, found2)
 }
-func Test_OnUpstreamRTCP(t *testing.T) {
+func Test_OnUpstreamStat(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
@@ -243,13 +306,36 @@ func Test_OnUpstreamRTCP(t *testing.T) {
 	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil)
 
 	// do
-
-	stat1 := &livekit.AnalyticsStat{NackCount: 1, PliCount: 1, FirCount: 1, Jitter: 5, PacketLost: 3, TotalPackets: 1, TotalBytes: 1}
+	stat1 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   1,
+				TotalPrimaryPackets: 1,
+				TotalPacketsLost:    3,
+				TotalNacks:          1,
+				TotalPlis:           1,
+				TotalFirs:           1,
+				Jitter:              5,
+			},
+		},
+	}
 	trackID := livekit.TrackID("trackID")
 
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat1)
 
-	stat2 := &livekit.AnalyticsStat{Jitter: 2, PacketLost: 4, TotalPackets: 2, TotalBytes: 2}
+	stat2 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   2,
+				TotalPrimaryPackets: 2,
+				TotalPacketsLost:    4,
+				TotalNacks:          1,
+				TotalPlis:           1,
+				TotalFirs:           1,
+				Jitter:              2,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat2)
 	fixture.sut.SendAnalytics()
 
@@ -258,12 +344,12 @@ func Test_OnUpstreamRTCP(t *testing.T) {
 	_, stats := fixture.analytics.SendStatsArgsForCall(0)
 	require.Equal(t, 1, len(stats))
 	require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind)
-	require.Equal(t, 1, int(stats[0].NackCount))
-	require.Equal(t, 1, int(stats[0].PliCount))
-	require.Equal(t, 1, int(stats[0].FirCount))
-	require.Equal(t, 0, int(stats[0].Rtt))        // TODO: test for RTT
-	require.Equal(t, 5, int(stats[0].Jitter))     // max of jitter, see list of rtcp.ReceptionReport above
-	require.Equal(t, 4, int(stats[0].PacketLost)) // last reported packets lost, see list of rtcp.ReceptionReport above
+	require.Equal(t, 1, int(stats[0].Streams[0].TotalNacks))
+	require.Equal(t, 1, int(stats[0].Streams[0].TotalPlis))
+	require.Equal(t, 1, int(stats[0].Streams[0].TotalFirs))
+	require.Equal(t, 0, int(stats[0].Streams[0].Rtt))              // TODO: test for RTT
+	require.Equal(t, 5, int(stats[0].Streams[0].Jitter))           // max of jitter, see list of rtcp.ReceptionReport above
+	require.Equal(t, 4, int(stats[0].Streams[0].TotalPacketsLost)) // last reported packets lost, see list of rtcp.ReceptionReport above
 	require.Equal(t, string(trackID), stats[0].TrackId)
 }
 
@@ -282,16 +368,40 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
 	trackID1 := livekit.TrackID("trackID1")
 	trackID2 := livekit.TrackID("trackID2")
 
-	stat1 := &livekit.AnalyticsStat{TotalBytes: uint64(totalBytes), TotalPackets: uint64(totalPackets)}
+	stat1 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   uint64(totalBytes),
+				TotalPrimaryPackets: uint32(totalPackets),
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID1, stat1)
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID2, stat1) // using same buffer is not correct but for test it is fine
 
 	// do
 	totalBytes++
 	totalPackets++
-	stat2 := &livekit.AnalyticsStat{NackCount: 1, TotalBytes: uint64(totalBytes), TotalPackets: uint64(totalPackets)}
+	stat2 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   uint64(totalBytes),
+				TotalPrimaryPackets: uint32(totalPackets),
+				TotalNacks:          1,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID1, stat2)
-	stat3 := &livekit.AnalyticsStat{FirCount: 1, TotalBytes: uint64(totalBytes), TotalPackets: uint64(totalPackets)}
+
+	stat3 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   uint64(totalBytes),
+				TotalPrimaryPackets: uint32(totalPackets),
+				TotalFirs:           1,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID2, stat3)
 	fixture.sut.SendAnalytics()
 
@@ -306,30 +416,22 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
 		if livekit.TrackID(sentStat.TrackId) == trackID1 {
 			found1 = true
 			require.Equal(t, livekit.StreamType_UPSTREAM, sentStat.Kind)
-			require.Equal(t, 1, int(sentStat.NackCount)) // see pkts1 above
+			require.Equal(t, 1, int(sentStat.Streams[0].TotalNacks)) // see pkts1 above
 		} else if livekit.TrackID(sentStat.TrackId) == trackID2 {
 			found2 = true
 			require.Equal(t, livekit.StreamType_UPSTREAM, sentStat.Kind)
-			require.Equal(t, 1, int(sentStat.FirCount)) // see pkts2 above
+			require.Equal(t, 1, int(sentStat.Streams[0].TotalFirs)) // see pkts2 above
 		}
-		require.Equal(t, totalBytes, int(sentStat.TotalBytes))
-		require.Equal(t, totalPackets, int(sentStat.TotalPackets))
+		require.Equal(t, totalBytes, int(sentStat.Streams[0].TotalPrimaryBytes))
+		require.Equal(t, totalPackets, int(sentStat.Streams[0].TotalPrimaryPackets))
 	}
 	require.True(t, found1)
 	require.True(t, found2)
 
-	// remove 1 track
+	// remove 1 track - track stats were flushed above, so no more calls to SendStats
 	fixture.sut.TrackUnpublished(context.Background(), partSID, &livekit.TrackInfo{Sid: string(trackID2)}, 0)
 	fixture.sut.SendAnalytics()
-	require.Equal(t, 2, fixture.analytics.SendStatsCallCount())
-	_, stats = fixture.analytics.SendStatsArgsForCall(1)
-	require.Equal(t, 2, len(stats)) // still 2 tracks, next call won't contain 1 track
-
-	// now only 1 track stats remaining
-	fixture.sut.SendAnalytics()
-	require.Equal(t, 3, fixture.analytics.SendStatsCallCount())
-	_, stats = fixture.analytics.SendStatsArgsForCall(2)
-	require.Equal(t, 1, len(stats)) // now only 1 track remaining
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
 }
 
 func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) {
@@ -344,8 +446,8 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) {
 	// do
 	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo)
 
-	// test
-	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	// should not be called if there are not track stats
+	require.Equal(t, 0, fixture.analytics.SendStatsCallCount())
 }
 
 func Test_AddUpTrack(t *testing.T) {
@@ -361,7 +463,14 @@ func Test_AddUpTrack(t *testing.T) {
 	var totalBytes uint64 = 3
 	var totalPackets uint32 = 3
 
-	stat := &livekit.AnalyticsStat{TotalPackets: uint64(totalPackets), TotalBytes: totalBytes}
+	stat := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   totalBytes,
+				TotalPrimaryPackets: totalPackets,
+			},
+		},
+	}
 	trackID := livekit.TrackID("trackID")
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat)
 	fixture.sut.SendAnalytics()
@@ -371,8 +480,8 @@ func Test_AddUpTrack(t *testing.T) {
 	_, stats := fixture.analytics.SendStatsArgsForCall(0)
 	require.Equal(t, 1, len(stats))
 	require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind)
-	require.Equal(t, totalBytes, stats[0].TotalBytes)
-	require.Equal(t, totalPackets, uint32(stats[0].TotalPackets))
+	require.Equal(t, totalBytes, stats[0].Streams[0].TotalPrimaryBytes)
+	require.Equal(t, totalPackets, uint32(stats[0].Streams[0].TotalPrimaryPackets))
 	require.Equal(t, string(trackID), stats[0].TrackId)
 }
 
@@ -387,10 +496,24 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) {
 	// do
 	trackID := livekit.TrackID("trackID")
 
-	stat1 := &livekit.AnalyticsStat{TotalBytes: 1, TotalPackets: 1}
+	stat1 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   1,
+				TotalPrimaryPackets: 1,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat1)
 
-	stat2 := &livekit.AnalyticsStat{TotalPackets: 2, TotalBytes: 2}
+	stat2 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   2,
+				TotalPrimaryPackets: 2,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat2)
 	fixture.sut.SendAnalytics()
 	// test
@@ -398,8 +521,8 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) {
 	_, stats := fixture.analytics.SendStatsArgsForCall(0)
 	require.Equal(t, 1, len(stats))
 	require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind)
-	require.Equal(t, stat2.TotalBytes, stats[0].TotalBytes)
-	require.Equal(t, stat2.TotalPackets, stats[0].TotalPackets)
+	require.Equal(t, stat2.Streams[0].TotalPrimaryBytes, stats[0].Streams[0].TotalPrimaryBytes)
+	require.Equal(t, stat2.Streams[0].TotalPrimaryPackets, stats[0].Streams[0].TotalPrimaryPackets)
 	require.Equal(t, string(trackID), stats[0].TrackId)
 }
 
@@ -414,10 +537,24 @@ func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) {
 
 	// do
 	// upstream bytes
-	stat1 := &livekit.AnalyticsStat{TotalPackets: 3, TotalBytes: 3}
+	stat1 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   3,
+				TotalPrimaryPackets: 3,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, "trackID", stat1)
 	// downstream bytes
-	stat2 := &livekit.AnalyticsStat{TotalPackets: 1, TotalBytes: 1}
+	stat2 := &livekit.AnalyticsStat{
+		Streams: []*livekit.AnalyticsStream{
+			{
+				TotalPrimaryBytes:   1,
+				TotalPrimaryPackets: 1,
+			},
+		},
+	}
 	fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, "trackID1", stat2)
 	fixture.sut.SendAnalytics()