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

pkg/rtc/participant.go

@@ -836,7 +836,7 @@ func (p *ParticipantImpl) SetMigrateState(s types.MigrateState) {
 		return
 	}
 
-	p.params.Logger.Infow("SetMigrateState", "state", s)
+	p.params.Logger.Debugw("SetMigrateState", "state", s)
 	p.migrateState.Store(s)
 	p.dirty.Store(true)
 
@@ -1536,7 +1536,7 @@ func (p *ParticipantImpl) onSubscribedMaxQualityChange(trackID livekit.TrackID,
 		)
 	}
 
-	p.pubLogger.Infow(
+	p.pubLogger.Debugw(
 		"sending max subscribed quality",
 		"trackID", trackID,
 		"qualities", subscribedQualities,

pkg/rtc/room.go

@@ -1330,7 +1330,8 @@ func BroadcastDataPacketForRoom(r types.Room, source types.LocalParticipant, dp
 
 	utils.ParallelExec(destParticipants, dataForwardLoadBalanceThreshold, 1, func(op types.LocalParticipant) {
 		err := op.SendDataPacket(dp, dpData)
-		if err != nil && !errors.Is(err, io.ErrClosedPipe) && !errors.Is(err, psrpc.Canceled) {
+		if err != nil && !errors.Is(err, io.ErrClosedPipe) && !errors.Is(err, psrpc.Canceled) &&
+			!errors.Is(err, psrpc.ErrStreamClosed) {
 			op.GetLogger().Infow("send data packet error", "error", err)
 		}
 	})

pkg/service/rtcservice.go

@@ -371,7 +371,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 					websocket.CloseNormalClosure,
 					websocket.CloseNoStatusReceived,
 				) {
-				pLogger.Infow("exit ws read loop for closed connection", "connID", cr.ConnectionID, "wsError", err)
+				pLogger.Debugw("exit ws read loop for closed connection", "connID", cr.ConnectionID, "wsError", err)
 			} else {
 				pLogger.Errorw("error reading from websocket", err, "connID", cr.ConnectionID)
 			}

pkg/sfu/buffer/rtpstats.go

@@ -1677,7 +1677,7 @@ func (r *RTPStats) getAndResetSnapshot(snapshotId uint32, override bool) (*Snaps
 
 	then := r.snapshots[snapshotId]
 	if then == nil {
-		extStartSN := r.sequenceNumber.GetExtendedHighest()
+		extStartSN := r.sequenceNumber.GetExtendedStart()
 		then = &Snapshot{
 			startTime:            r.startTime,
 			extStartSN:           extStartSN,

pkg/sfu/streamallocator/probe_controller.go

@@ -154,7 +154,7 @@ func (p *ProbeController) MaybeFinalizeProbe(
 			queueWait = p.params.Config.SettleWaitMax
 		}
 		p.probeEndTime = p.lastProbeStartTime.Add(queueWait + p.doneProbeClusterInfo.Duration)
-		p.params.Logger.Infow(
+		p.params.Logger.Debugw(
 			"setting probe end time",
 			"probeClusterId", p.probeClusterId,
 			"expectedDuration", expectedDuration,

pkg/sfu/streamallocator/streamallocator.go

@@ -991,7 +991,7 @@ func (s *StreamAllocator) onProbeDone(isNotFailing bool, isGoalReached bool) {
 	//
 	channelObserverString := s.channelObserver.ToString()
 	s.channelObserver = s.newChannelObserverNonProbe()
-	s.params.Logger.Infow(
+	s.params.Logger.Debugw(
 		"probe done",
 		"isNotFailing", isNotFailing,
 		"isGoalReached", isGoalReached,