Fix infinitate loop possible in sample builder

Active becomes "empty" after consuming the packets, but instead of the
"filled" getting purged by "purgeConsumedBuffers", it doesn't because
active is now empty and empty buffers cannot cause any purging. So the
solution is to purge the "consumed" area first, then "active".

Fixes #1810

Author: robin-raymond

pkg/media/samplebuilder/samplebuilder.go

@@ -112,7 +112,23 @@ func (s *SampleBuilder) releasePacket(i uint16) {
 // purgeConsumedBuffers clears all buffers that have already been consumed by
 // popping.
 func (s *SampleBuilder) purgeConsumedBuffers() {
-	for s.active.compare(s.filled.head) == slCompareBefore && s.filled.hasData() {
+	s.purgeConsumedLocation(s.active, false)
+}
+
+// purgeConsumedLocation clears all buffers that have already been consumed
+// during a sample building method.
+func (s *SampleBuilder) purgeConsumedLocation(consume sampleSequenceLocation, forceConsume bool) {
+	if !s.filled.hasData() {
+		return
+	}
+
+	switch consume.compare(s.filled.head) {
+	case slCompareInside:
+		if !forceConsume {
+			break
+		}
+		fallthrough
+	case slCompareBefore:
 		s.releasePacket(s.filled.head)
 		s.filled.head++
 	}
@@ -231,15 +247,16 @@ func (s *SampleBuilder) buildSample(purgingBuffers bool) *media.Sample {
 	if s.partitionHeadChecker != nil {
 		if !s.partitionHeadChecker.IsPartitionHead(s.buffer[consume.head].Payload) {
 			s.droppedPackets += consume.count()
+			s.purgeConsumedLocation(consume, true)
 			s.purgeConsumedBuffers()
 			return nil
 		}
 	}
 
 	// merge all the buffers into a sample
 	data := []byte{}
-	for ; consume.head != consume.tail; consume.head++ {
-		p, err := s.depacketizer.Unmarshal(s.buffer[consume.head].Payload)
+	for i := consume.head; i != consume.tail; i++ {
+		p, err := s.depacketizer.Unmarshal(s.buffer[i].Payload)
 		if err != nil {
 			return nil
 		}
@@ -259,6 +276,7 @@ func (s *SampleBuilder) buildSample(purgingBuffers bool) *media.Sample {
 	s.preparedSamples[s.prepared.tail] = sample
 	s.prepared.tail++
 
+	s.purgeConsumedLocation(consume, true)
 	s.purgeConsumedBuffers()
 
 	return sample

pkg/media/samplebuilder/samplebuilder_test.go

@@ -328,6 +328,20 @@ func TestSampleBuilderCleanReference(t *testing.T) {
 	}
 }
 
+func TestSampleBuilderPushMaxZero(t *testing.T) {
+	// Test packets released via 'maxLate' of zero.
+	pkts := []rtp.Packet{
+		{Header: rtp.Header{SequenceNumber: 0, Timestamp: 0, Marker: true}, Payload: []byte{0x01}},
+	}
+	s := New(0, &fakeDepacketizer{}, 1, WithPartitionHeadChecker(
+		&fakePartitionHeadChecker{headBytes: []byte{0x01}},
+	))
+	s.Push(&pkts[0])
+	if sample := s.Pop(); sample == nil {
+		t.Error("Should expect a popped sample")
+	}
+}
+
 func TestSampleBuilderWithPacketReleaseHandler(t *testing.T) {
 	var released []*rtp.Packet
 	fakePacketReleaseHandler := func(p *rtp.Packet) {