Connection quality LOST only if RTCP is also not available. (#2670)

* Connection quality LOST only if RTCP is also not available.

It is possible that sender stops all layers of video due to some
constraint (CPU or bandwidth). Packet reception going dry due to
that should not trigger `LOST` quality.

Add last received RTCP time also to distinguish the case
of real `LOST` and sender stopping traffic.

Some bits to watch for
- With audio, RTCP reports could be more than 5 seconds apart (5 seconds
  is the default interval for connection quality scorer), but audio
  senders usually send silence packets even when there is no input.
  So audio completely stopping can be considered `LOST`.
- With video, have to observe if all clients continue to send RTCP even
  if all layers are stopped.
- RTCP bandwidth is not supposed to exceed the primary stream bandwidth.
  libwebrtc calculates that and spaces out RTCP reports accordingly.
  That is the reason why audio reports are that far apart. If a video
  stream is encoded at a very low bit rate, it could also be sending
  RTCP rarely. So, there is the case of LOST being indistinguishable
  from sender stopping all layers. But, this should be a rare case.

* typo

pkg/sfu/buffer/buffer.go

@@ -976,6 +976,17 @@ func (b *Buffer) GetDeltaStats() *StreamStatsWithLayers {
 	}
 }
 
+func (b *Buffer) GetLastSenderReportTime() time.Time {
+	b.RLock()
+	defer b.RUnlock()
+
+	if b.rtpStats == nil {
+		return time.Time{}
+	}
+
+	return b.rtpStats.LastSenderReportTime()
+}
+
 func (b *Buffer) GetAudioLevel() (float64, bool) {
 	b.RLock()
 	defer b.RUnlock()

pkg/sfu/buffer/rtpstats_receiver.go

@@ -458,6 +458,17 @@ func (r *RTPStatsReceiver) GetRtcpSenderReportData() *RTCPSenderReportData {
 	return &srNewestCopy
 }
 
+func (r *RTPStatsReceiver) LastSenderReportTime() time.Time {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+
+	if r.srNewest != nil {
+		return r.srNewest.At
+	}
+
+	return time.Time{}
+}
+
 func (r *RTPStatsReceiver) GetRtcpReceptionReport(ssrc uint32, proxyFracLost uint8, snapshotID uint32) *rtcp.ReceptionReport {
 	r.lock.Lock()
 	defer r.lock.Unlock()

pkg/sfu/connectionquality/connectionstats.go

@@ -37,6 +37,7 @@ const (
 
 type ConnectionStatsReceiverProvider interface {
 	GetDeltaStats() map[uint32]*buffer.StreamStatsWithLayers
+	GetLastSenderReportTime() time.Time
 }
 
 type ConnectionStatsSenderProvider interface {
@@ -202,7 +203,7 @@ func (cs *ConnectionStats) GetScoreAndQuality() (float32, livekit.ConnectionQual
 	return cs.scorer.GetMOSAndQuality()
 }
 
-func (cs *ConnectionStats) updateScoreWithAggregate(agg *buffer.RTPDeltaInfo, at time.Time) float32 {
+func (cs *ConnectionStats) updateScoreWithAggregate(agg *buffer.RTPDeltaInfo, lastRTCPAt time.Time, at time.Time) float32 {
 	var stat windowStat
 	if agg != nil {
 		stat.startedAt = agg.StartTime
@@ -214,6 +215,8 @@ func (cs *ConnectionStats) updateScoreWithAggregate(agg *buffer.RTPDeltaInfo, at
 		stat.bytes = agg.Bytes - agg.HeaderBytes // only use media payload size
 		stat.rttMax = agg.RttMax
 		stat.jitterMax = agg.JitterMax
+
+		stat.lastRTCPAt = lastRTCPAt
 	}
 	if at.IsZero() {
 		cs.scorer.Update(&stat)
@@ -246,7 +249,7 @@ func (cs *ConnectionStats) updateScoreFromReceiverReport(at time.Time) (float32,
 		}
 		if time.Since(marker) > noReceiverReportTooLongThreshold {
 			// have not received receiver report for a long time when streaming, run with nil stat
-			return cs.updateScoreWithAggregate(nil, at), nil
+			return cs.updateScoreWithAggregate(nil, time.Time{}, at), nil
 		}
 
 		// wait for receiver report, return current score
@@ -266,7 +269,7 @@ func (cs *ConnectionStats) updateScoreFromReceiverReport(at time.Time) (float32,
 	if streamingStartedAt.After(agg.StartTime) {
 		agg.StartTime = streamingStartedAt
 	}
-	return cs.updateScoreWithAggregate(agg, at), streams
+	return cs.updateScoreWithAggregate(agg, time.Time{}, at), streams
 }
 
 func (cs *ConnectionStats) updateScoreAt(at time.Time) (float32, map[uint32]*buffer.StreamStatsWithLayers) {
@@ -290,7 +293,7 @@ func (cs *ConnectionStats) updateScoreAt(at time.Time) (float32, map[uint32]*buf
 		deltaInfoList = append(deltaInfoList, s.RTPStats)
 	}
 	agg := buffer.AggregateRTPDeltaInfo(deltaInfoList)
-	return cs.updateScoreWithAggregate(agg, at), streams
+	return cs.updateScoreWithAggregate(agg, cs.params.ReceiverProvider.GetLastSenderReportTime(), at), streams
 }
 
 func (cs *ConnectionStats) updateStreamingStart(at time.Time) time.Time {

pkg/sfu/connectionquality/connectionstats_test.go

@@ -29,7 +29,8 @@ import (
 // -----------------------------------------------
 
 type testReceiverProvider struct {
-	streams map[uint32]*buffer.StreamStatsWithLayers
+	streams              map[uint32]*buffer.StreamStatsWithLayers
+	lastSenderReportTime time.Time
 }
 
 func newTestReceiverProvider() *testReceiverProvider {
@@ -44,6 +45,14 @@ func (trp *testReceiverProvider) GetDeltaStats() map[uint32]*buffer.StreamStatsW
 	return trp.streams
 }
 
+func (trp *testReceiverProvider) setLastSenderReportTime(at time.Time) {
+	trp.lastSenderReportTime = at
+}
+
+func (trp *testReceiverProvider) GetLastSenderReportTime() time.Time {
+	return trp.lastSenderReportTime
+}
+
 // -----------------------------------------------
 
 func TestConnectionQuality(t *testing.T) {
@@ -232,7 +241,7 @@ func TestConnectionQuality(t *testing.T) {
 		require.Greater(t, float32(4.6), mos)
 		require.Equal(t, livekit.ConnectionQuality_EXCELLENT, quality)
 
-		// unmute at time so that next window does not satisfy the unmute time threshold.
+		// unmute at specific time to ensure next window does not satisfy the unmute time threshold.
 		// that means even if the next update has 0 packets, it should hold state and stay at EXCELLENT quality
 		cs.UpdateMuteAt(false, now.Add(3*time.Second))
 
@@ -250,7 +259,8 @@ func TestConnectionQuality(t *testing.T) {
 		require.Greater(t, float32(4.6), mos)
 		require.Equal(t, livekit.ConnectionQuality_EXCELLENT, quality)
 
-		// next update with no packets should knock quality down to LOST
+		// next update with no packets,
+		// but last RTCP is not set, should knock quality down to POOR
 		now = now.Add(duration)
 		trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
 			1: {
@@ -264,13 +274,46 @@ func TestConnectionQuality(t *testing.T) {
 		cs.updateScoreAt(now.Add(duration))
 		mos, quality = cs.GetScoreAndQuality()
 		require.Greater(t, float32(2.1), mos)
+		require.Equal(t, livekit.ConnectionQuality_POOR, quality)
+
+		// another dry spell, but last RTCP is not stale, should keep quality at POOR
+		now = now.Add(duration)
+		trp.setLastSenderReportTime(now.Add(time.Second))
+		trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
+			1: {
+				RTPStats: &buffer.RTPDeltaInfo{
+					StartTime: now,
+					EndTime:   now.Add(duration),
+					Packets:   0,
+				},
+			},
+		})
+		cs.updateScoreAt(now.Add(duration))
+		mos, quality = cs.GetScoreAndQuality()
+		require.Greater(t, float32(2.1), mos)
+		require.Equal(t, livekit.ConnectionQuality_POOR, quality)
+
+		// yet another dry spell, but last RTCP is stale, should knock down quality at LOST
+		now = now.Add(duration)
+		trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
+			1: {
+				RTPStats: &buffer.RTPDeltaInfo{
+					StartTime: now,
+					EndTime:   now.Add(duration),
+					Packets:   0,
+				},
+			},
+		})
+		cs.updateScoreAt(now.Add(duration))
+		mos, quality = cs.GetScoreAndQuality()
+		require.Greater(t, float32(1.3), mos)
 		require.Equal(t, livekit.ConnectionQuality_LOST, quality)
 
 		// mute when LOST should not bump up score/quality
 		now = now.Add(duration)
 		cs.UpdateMuteAt(true, now.Add(1*time.Second))
 		mos, quality = cs.GetScoreAndQuality()
-		require.Greater(t, float32(2.1), mos)
+		require.Greater(t, float32(1.3), mos)
 		require.Equal(t, livekit.ConnectionQuality_LOST, quality)
 
 		// unmute and send packets to bring quality back up

pkg/sfu/connectionquality/scorer.go

@@ -61,6 +61,7 @@ type windowStat struct {
 	bytes             uint64
 	rttMax            uint32
 	jitterMax         float64
+	lastRTCPAt        time.Time
 }
 
 func (w *windowStat) calculatePacketScore(plw float64, includeRTT bool, includeJitter bool) float64 {
@@ -147,7 +148,7 @@ func (w *windowStat) calculateBitrateScore(expectedBitrate int64, isEnabled bool
 }
 
 func (w *windowStat) String() string {
-	return fmt.Sprintf("start: %+v, dur: %+v, pe: %d, pl: %d, pm: %d, pooo: %d, b: %d, rtt: %d, jitter: %0.2f",
+	return fmt.Sprintf("start: %+v, dur: %+v, pe: %d, pl: %d, pm: %d, pooo: %d, b: %d, rtt: %d, jitter: %0.2f, lastRTCP: %+v",
 		w.startedAt,
 		w.duration,
 		w.packetsExpected,
@@ -157,6 +158,7 @@ func (w *windowStat) String() string {
 		w.bytes,
 		w.rttMax,
 		w.jitterMax,
+		w.lastRTCPAt,
 	)
 }
 
@@ -174,6 +176,7 @@ func (w *windowStat) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	e.AddUint64("bytes", w.bytes)
 	e.AddUint32("rttMax", w.rttMax)
 	e.AddFloat64("jitterMax", w.jitterMax)
+	e.AddTime("lastRTCPAt", w.lastRTCPAt)
 	return nil
 }
 
@@ -393,8 +396,13 @@ func (q *qualityScorer) updateAtLocked(stat *windowStat, at time.Time) {
 	reason := "none"
 	var score float64
 	if stat.packetsExpected == 0 {
-		reason = "dry"
-		score = qualityTransitionScore[livekit.ConnectionQuality_LOST]
+		if !stat.lastRTCPAt.IsZero() && at.Sub(stat.lastRTCPAt) > stat.duration {
+			reason = "dry"
+			score = qualityTransitionScore[livekit.ConnectionQuality_LOST]
+		} else {
+			reason = "rtcp"
+			score = qualityTransitionScore[livekit.ConnectionQuality_POOR]
+		}
 	} else {
 		packetScore := stat.calculatePacketScore(plw, q.params.IncludeRTT, q.params.IncludeJitter)
 		bitrateScore := stat.calculateBitrateScore(expectedBitrate, q.params.EnableBitrateScore)

pkg/sfu/receiver.go

@@ -647,6 +647,25 @@ func (w *WebRTCReceiver) GetDeltaStats() map[uint32]*buffer.StreamStatsWithLayer
 	return deltaStats
 }
 
+func (w *WebRTCReceiver) GetLastSenderReportTime() time.Time {
+	w.bufferMu.RLock()
+	defer w.bufferMu.RUnlock()
+
+	latestSRTime := time.Time{}
+	for _, buff := range w.buffers {
+		if buff == nil {
+			continue
+		}
+
+		srAt := buff.GetLastSenderReportTime()
+		if srAt.After(latestSRTime) {
+			latestSRTime = srAt
+		}
+	}
+
+	return latestSRTime
+}
+
 func (w *WebRTCReceiver) forwardRTP(layer int32) {
 	pktBuf := make([]byte, bucket.MaxPktSize)
 	tracker := w.streamTrackerManager.GetTracker(layer)