Slight sequencer tweaks. (#2184)

The buffer is not for padding packets. So, calculate
adjusted sequence numbers before comparing against size.

Also, it is possible that invalidated slot is accessed
due to not being able to exclude padding range. This was
causing time stamp reset to 0. Will remove the error log
after this goes out and the condition does not show up
for a few days.

pkg/sfu/buffer/rtpstats_receiver.go

@@ -154,7 +154,7 @@ func (r *RTPStatsReceiver) Update(
 				return
 			}
 		}
-		if -gapSN >= cNumSequenceNumbers {
+		if -gapSN >= cNumSequenceNumbers/2 {
 			r.logger.Warnw(
 				"large sequence number gap negative", nil,
 				"extStartSN", r.sequenceNumber.GetExtendedStart(),
@@ -226,7 +226,7 @@ func (r *RTPStatsReceiver) Update(
 		flowState.ExtSequenceNumber = resSN.ExtendedVal
 		flowState.ExtTimestamp = resTS.ExtendedVal
 	} else { // in-order
-		if gapSN >= cNumSequenceNumbers {
+		if gapSN >= cNumSequenceNumbers/2 {
 			r.logger.Warnw(
 				"large sequence number gap", nil,
 				"extStartSN", r.sequenceNumber.GetExtendedStart(),

pkg/sfu/buffer/rtpstats_sender.go

@@ -286,7 +286,7 @@ func (r *RTPStatsSender) Update(
 			// do not start on a padding only packet
 			return
 		}
-		if -gapSN >= cNumSequenceNumbers {
+		if -gapSN >= cNumSequenceNumbers/2 {
 			r.logger.Warnw(
 				"large sequence number gap negative", nil,
 				"extStartSN", r.extStartSN,
@@ -352,7 +352,7 @@ func (r *RTPStatsSender) Update(
 			r.setSnInfo(extSequenceNumber, r.extHighestSN, uint16(pktSize), uint8(hdrSize), uint16(payloadSize), marker, true)
 		}
 	} else { // in-order
-		if gapSN >= cNumSequenceNumbers {
+		if gapSN >= cNumSequenceNumbers/2 {
 			r.logger.Warnw(
 				"large sequence number gap", nil,
 				"extStartSN", r.extStartSN,
@@ -399,8 +399,12 @@ func (r *RTPStatsSender) Update(
 				"invalid start timestamp", nil,
 				"snBefore", r.extStartSN,
 				"snAfter", extSequenceNumber,
+				"snHighest", r.extHighestSN,
 				"tsBefore", r.extStartTS,
 				"tsAfter", extTimestamp,
+				"tsHighest", r.extHighestTS,
+				"firstTime", r.firstTime.String(),
+				"startTime", r.startTime.String(),
 			)
 		}
 		if extTimestamp != 0 {

pkg/sfu/sequencer.go

@@ -84,6 +84,7 @@ type sequencer struct {
 	size         int
 	startTime    int64
 	initialized  bool
+	extStartSN   uint64
 	extHighestSN uint64
 	snOffset     uint64
 	extHighestTS uint64
@@ -133,46 +134,61 @@ func (s *sequencer) push(
 
 	if !s.initialized {
 		s.initialized = true
-		s.extHighestSN = extModifiedSN - 1
+		s.extStartSN = extModifiedSN
+		s.extHighestSN = extModifiedSN
 		s.extHighestTS = extModifiedTS
 		s.updateSNOffset()
 	}
 
-	snOffset := s.snOffset
-	diff := int64(extModifiedSN - s.extHighestSN)
-	if diff >= 0 {
-		s.extHighestSN = extModifiedSN
-	} else {
-		if diff < -int64(s.size) {
-			s.logger.Warnw(
-				"old packet, cannot be sequenced", nil,
-				"extHighestSN", s.extHighestSN,
-				"extIncomingSN", extIncomingSN,
-				"extModifiedSN", extModifiedSN,
-			)
-			return
-		}
+	if extModifiedSN < s.extStartSN {
+		// old packet, should not happen
+		return
+	}
 
+	extHighestSNAdjusted := s.extHighestSN - s.snOffset
+	extModifiedSNAdjusted := extModifiedSN - s.snOffset
+	if extModifiedSN < s.extHighestSN {
 		if s.snRangeMap != nil {
-			var err error
-			snOffset, err = s.snRangeMap.GetValue(extModifiedSN)
+			snOffset, err := s.snRangeMap.GetValue(extModifiedSN)
 			if err != nil {
 				s.logger.Errorw(
 					"could not get sequence number offset", err,
+					"extStartSN", s.extStartSN,
 					"extHighestSN", s.extHighestSN,
 					"extIncomingSN", extIncomingSN,
 					"extModifiedSN", extModifiedSN,
+					"snOffset", s.snOffset,
 				)
 				return
 			}
+
+			extModifiedSNAdjusted = extModifiedSN - snOffset
+		}
+	}
+
+	if int64(extModifiedSNAdjusted-extHighestSNAdjusted) <= -int64(s.size) {
+		s.logger.Warnw(
+			"old packet, cannot be sequenced", nil,
+			"extHighestSN", s.extHighestSN,
+			"extIncomingSN", extIncomingSN,
+			"extModifiedSN", extModifiedSN,
+		)
+		return
+	}
+
+	// invalidate missing sequence numbers
+	if extModifiedSNAdjusted > extHighestSNAdjusted {
+		numInvalidated := 0
+		for esn := extHighestSNAdjusted + 1; esn != extModifiedSNAdjusted; esn++ {
+			s.invalidateSlot(int(esn % uint64(s.size)))
+			numInvalidated++
+			if numInvalidated >= s.size {
+				break
+			}
 		}
 	}
 
-	if int64(extModifiedTS-s.extHighestTS) >= 0 {
-		s.extHighestTS = extModifiedTS
-	}
-
-	slot := (extModifiedSN - snOffset) % uint64(s.size)
+	slot := extModifiedSNAdjusted % uint64(s.size)
 	s.meta[slot] = packetMeta{
 		sourceSeqNo: uint16(extIncomingSN),
 		targetSeqNo: uint16(extModifiedSN),
@@ -183,6 +199,13 @@ func (s *sequencer) push(
 		ddBytes:     append([]byte{}, ddBytes...),
 		lastNack:    s.getRefTime(packetTime), // delay retransmissions after the original transmission
 	}
+
+	if extModifiedSN > s.extHighestSN {
+		s.extHighestSN = extModifiedSN
+	}
+	if extModifiedTS > s.extHighestTS {
+		s.extHighestTS = extModifiedTS
+	}
 }
 
 func (s *sequencer) pushPadding(extStartSNInclusive uint64, extEndSNInclusive uint64) {
@@ -223,10 +246,7 @@ func (s *sequencer) pushPadding(extStartSNInclusive uint64, extEndSNInclusive ui
 			}
 
 			slot := (sn - snOffset) % uint64(s.size)
-			s.meta[slot] = packetMeta{
-				sourceSeqNo: 0,
-				targetSeqNo: 0,
-			}
+			s.invalidateSlot(int(slot))
 		}
 		return
 	}
@@ -244,6 +264,10 @@ func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta {
 	s.Lock()
 	defer s.Unlock()
 
+	if !s.initialized {
+		return nil
+	}
+
 	snOffset := uint64(0)
 	var err error
 	extPacketMetas := make([]extPacketMeta, 0, len(seqNo))
@@ -263,11 +287,6 @@ func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta {
 			extSN -= (1 << 16)
 		}
 
-		if s.extHighestSN-extSN >= uint64(s.size) {
-			// too old
-			continue
-		}
-
 		if s.snRangeMap != nil {
 			snOffset, err = s.snRangeMap.GetValue(extSN)
 			if err != nil {
@@ -276,9 +295,17 @@ func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta {
 			}
 		}
 
-		slot := (extSN - snOffset) % uint64(s.size)
+		extSNAdjusted := extSN - snOffset
+		extHighestSNAdjusted := s.extHighestSN - s.snOffset
+		if extHighestSNAdjusted-extSNAdjusted >= uint64(s.size) {
+			// too old
+			continue
+		}
+
+		slot := extSNAdjusted % uint64(s.size)
 		meta := &s.meta[slot]
-		if meta.targetSeqNo != sn {
+		if meta.targetSeqNo != sn || s.isInvalidSlot(int(slot)) {
+			// invalid slot access could happen if padding packets exclusion range could not be recorded
 			continue
 		}
 
@@ -320,3 +347,24 @@ func (s *sequencer) updateSNOffset() {
 	}
 	s.snOffset = snOffset
 }
+
+func (s *sequencer) invalidateSlot(slot int) {
+	if slot >= len(s.meta) {
+		return
+	}
+
+	s.meta[slot] = packetMeta{
+		sourceSeqNo: 0,
+		targetSeqNo: 0,
+		lastNack:    0,
+	}
+}
+
+func (s *sequencer) isInvalidSlot(slot int) bool {
+	if slot >= len(s.meta) {
+		return true
+	}
+
+	meta := &s.meta[slot]
+	return meta.sourceSeqNo == 0 && meta.targetSeqNo == 0 && meta.lastNack == 0
+}