Allow start streaming on an out-of-order packet. (#2971)

But, do not record first packet time on an out-of-order packet.
It so happens that packets get out-of-order a lot more across relay.
And it turns out with some H.264 stream, the first few packets of a key
frame are very small (may be SPS/PPS, haven't checked), they get
out-of-oder quite a lot, so much so a down track never starts even it
has 20 - 25 key frames have passed through.

pkg/sfu/buffer/rtpstats_sender.go

@@ -241,6 +241,7 @@ func (r *RTPStatsSender) Update(
 	hdrSize int,
 	payloadSize int,
 	paddingSize int,
+	isOutOfOrder bool,
 ) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
@@ -259,7 +260,6 @@ func (r *RTPStatsSender) Update(
 
 		r.startTime = time.Now()
 
-		r.firstTime = packetTime
 		r.highestTime = packetTime
 
 		r.extStartSN = extSequenceNumber
@@ -281,6 +281,12 @@ func (r *RTPStatsSender) Update(
 			"rtpStats", lockedRTPStatsSenderLogEncoder{r},
 		)
 	}
+	if !isOutOfOrder && r.firstTime == 0 {
+		// do not set first packet time if packet is out-of-order,
+		// as first packet time is used to calculate expected time stamp,
+		// using an out-of-order packet would skew that.
+		r.firstTime = packetTime
+	}
 
 	pktSize := uint64(hdrSize + payloadSize + paddingSize)
 	isDuplicate := false
@@ -595,7 +601,7 @@ func (r *RTPStatsSender) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt ui
 	r.lock.RLock()
 	defer r.lock.RUnlock()
 
-	if !r.initialized {
+	if r.firstTime == 0 {
 		err = errors.New("uninitialized")
 		return
 	}

pkg/sfu/downtrack.go

@@ -835,8 +835,7 @@ func (d *DownTrack) maxLayerNotifierWorker() {
 
 // WriteRTP writes an RTP Packet to the DownTrack
 func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
-	if !d.writable.Load() || (extPkt.IsOutOfOrder && !d.rtpStats.IsActive()) {
-		// do not start on an out-of-order packet
+	if !d.writable.Load() {
 		return nil
 	}
 
@@ -945,6 +944,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 			extSequenceNumber: tp.rtp.extSequenceNumber,
 			extTimestamp:      tp.rtp.extTimestamp,
 			isKeyFrame:        extPkt.KeyFrame,
+			isOutOfOrder:      extPkt.IsOutOfOrder,
 			tp:                &tp,
 		},
 	)
@@ -1947,6 +1947,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 				extSequenceNumber: epm.extSequenceNumber,
 				extTimestamp:      epm.extTimestamp,
 				isRTX:             true,
+				isOutOfOrder:      true,
 			},
 		)
 		d.pacer.Enqueue(pacer.Packet{
@@ -2212,6 +2213,7 @@ type sendPacketMetadata struct {
 	extTimestamp         uint64
 	isKeyFrame           bool
 	isRTX                bool
+	isOutOfOrder         bool
 	isPadding            bool
 	shouldDisableCounter bool
 	tp                   *TranslationParams
@@ -2233,11 +2235,13 @@ func (d *DownTrack) sendingPacket(hdr *rtp.Header, payloadSize int, spmd *sendPa
 	}
 
 	// update RTPStats
+	paddingSize := payloadSize
 	if spmd.isPadding {
-		d.rtpStats.Update(spmd.packetTime, spmd.extSequenceNumber, spmd.extTimestamp, hdr.Marker, hdrSize, 0, payloadSize)
+		payloadSize = 0
 	} else {
-		d.rtpStats.Update(spmd.packetTime, spmd.extSequenceNumber, spmd.extTimestamp, hdr.Marker, hdrSize, payloadSize, 0)
+		paddingSize = 0
 	}
+	d.rtpStats.Update(spmd.packetTime, spmd.extSequenceNumber, spmd.extTimestamp, hdr.Marker, hdrSize, payloadSize, paddingSize, spmd.isOutOfOrder)
 
 	if spmd.isKeyFrame {
 		d.isNACKThrottled.Store(false)