Repository: pion/interceptor
Pull request #198 (2023-07-27 03:01:21)

Partial rewrite of TWCC sender
The previous implementation of the TWCC sender interceptor had some inconsistencies with libwebrtc. Namely, if there were missing packets between the last packet in the previous feedback interval and the first received packet in the following feedback interval, then those packets were never included as missing in any feedback.

This is an issue because libwebrtc uses data about lost packets from TWCC feedback in their congestion controller. This means that bursts of loss could go unnoticed by libwebrtc, causing the application to possibly send more data than it would if the loss was properly reported.

Another minor difference was that feedback packets with a single packet are in fact valid according to the RFC, but we were only returning feedback with at least two packets. libwebrtc has a check for minimum feedback size, which has now been added to the unit tests.

Just about all of the rewrite here has been ported from libwebrtc code, so it should now match the behavior fairly well.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pion/interceptor/pull/198?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
Patch coverage: **`87.61%`** and project coverage change: **`+0.36%`** :tada:
> Comparison is base [(`9ded56b`)](https://app.codecov.io/gh/pion/interceptor/commit/9ded56baf425ed5633369d029b7138a587b2dc0e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) 79.41% compared to head [(`81af859`)](https://app.codecov.io/gh/pion/interceptor/pull/198?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) 79.77%.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master     #198      +/-   ##
==========================================
+ Coverage   79.41%   79.77%   +0.36%     
==========================================
  Files          67       68       +1     
  Lines        3367     3536     +169     
==========================================
+ Hits         2674     2821     +147     
- Misses        574      590      +16     
- Partials      119      125       +6     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `79.69% <87.61%> (+0.27%)` | :arrow_up: |
| wasm | `77.65% <86.19%> (+0.46%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Files Changed](https://app.codecov.io/gh/pion/interceptor/pull/198?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [pkg/twcc/twcc.go](https://app.codecov.io/gh/pion/interceptor/pull/198?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3R3Y2MvdHdjYy5nbw==) | `86.40% <71.11%> (-8.55%)` | :arrow_down: |
| [pkg/twcc/arrival\_time\_map.go](https://app.codecov.io/gh/pion/interceptor/pull/198?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3R3Y2MvYXJyaXZhbF90aW1lX21hcC5nbw==) | `100.00% <100.00%> (ø)` | |
| [pkg/twcc/sender\_interceptor.go](https://app.codecov.io/gh/pion/interceptor/pull/198?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3R3Y2Mvc2VuZGVyX2ludGVyY2VwdG9yLmdv) | `84.07% <100.00%> (ø)` | |

... and [2 files with indirect coverage changes](https://app.codecov.io/gh/pion/interceptor/pull/198/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/pion/interceptor/pull/198?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).

kcaffrey: I see #202 got merged, but I think this PR would actually replace it and do the same deferred sorting but in O(N) instead of O(N log N). This is because I walk through the sequence numbers in order to build feedback (note that since missing packets have to be added anyway, this adds no overhead). Simply keeping track of the start/end sequence numbers and storing arrival times in a circular buffer lets you do a simple for loop to get the sorted arrival times; no explicit sort needed.

@davidzhao @treyhakanson can you take a look and confirm that my logic above is correct, and that the proper way to resolve this change with #202 is to simply overwrite the changes?
kcaffrey: Ok, I have reverted #202 and rebased this branch on top of master. I left it as two separate commits to clarify the revert of #202.

Patch:
diff --git a/pkg/twcc/arrival_time_map.go b/pkg/twcc/arrival_time_map.go
new file mode 100644
index 0000000..abca1ff
--- /dev/null
+++ b/pkg/twcc/arrival_time_map.go
@@ -0,0 +1,192 @@
+// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+package twcc
+
+const (
+	minCapacity        = 128
+	maxNumberOfPackets = 1 << 15
+)
+
+// packetArrivalTimeMap is adapted from Chrome's implementation of TWCC, and keeps track
+// of the arrival times of packets. It is used by the TWCC interceptor to build feedback
+// packets.
+// See https://source.chromium.org/chromium/chromium/src/+/refs/heads/main:third_party/webrtc/modules/remote_bitrate_estimator/packet_arrival_map.h;drc=b5cd13bb6d5d157a5fbe3628b2dd1c1e106203c6
+type packetArrivalTimeMap struct {
+	// arrivalTimes is a circular buffer, where the packet with sequence number sn is stored
+	// in slot sn % len(arrivalTimes)
+	arrivalTimes []int64
+
+	// The unwrapped sequence numbers for the range of valid sequence numbers in arrivalTimes.
+	// beginSequenceNumber is inclusive, and endSequenceNumber is exclusive.
+	beginSequenceNumber, endSequenceNumber int64
+}
+
+// AddPacket records the fact that the packet with sequence number sequenceNumber arrived
+// at arrivalTime.
+func (m *packetArrivalTimeMap) AddPacket(sequenceNumber int64, arrivalTime int64) {
+	if m.arrivalTimes == nil {
+		// First packet
+		m.reallocate(minCapacity)
+		m.beginSequenceNumber = sequenceNumber
+		m.endSequenceNumber = sequenceNumber + 1
+		m.arrivalTimes[m.index(sequenceNumber)] = arrivalTime
+		return
+	}
+
+	if sequenceNumber >= m.beginSequenceNumber && sequenceNumber < m.endSequenceNumber {
+		// The packet is within the buffer, no need to resize.
+		m.arrivalTimes[m.index(sequenceNumber)] = arrivalTime
+		return
+	}
+
+	if sequenceNumber < m.beginSequenceNumber {
+		// The packet goes before the current buffer. Expand to add packet,
+		// but only if it fits within the maximum number of packets.
+		newSize := int(m.endSequenceNumber - sequenceNumber)
+		if newSize > maxNumberOfPackets {
+			// Don't expand the buffer back for this packet, as it would remove newer received
+			// packets.
+			return
+		}
+		m.adjustToSize(newSize)
+		m.arrivalTimes[m.index(sequenceNumber)] = arrivalTime
+		m.setNotReceived(sequenceNumber+1, m.beginSequenceNumber)
+		m.beginSequenceNumber = sequenceNumber
+		return
+	}
+
+	// The packet goes after the buffer.
+	newEndSequenceNumber := sequenceNumber + 1
+
+	if newEndSequenceNumber >= m.endSequenceNumber+maxNumberOfPackets {
+		// All old packets have to be removed.
+		m.beginSequenceNumber = sequenceNumber
+		m.endSequenceNumber = newEndSequenceNumber
+		m.arrivalTimes[m.index(sequenceNumber)] = arrivalTime
+		return
+	}
+
+	if m.beginSequenceNumber < newEndSequenceNumber-maxNumberOfPackets {
+		// Remove oldest entries.
+		m.beginSequenceNumber = newEndSequenceNumber - maxNumberOfPackets
+	}
+
+	m.adjustToSize(int(newEndSequenceNumber - m.beginSequenceNumber))
+
+	// Packets can be received out of order. If this isn't the next expected packet,
+	// add enough placeholders to fill the gap.
+	m.setNotReceived(m.endSequenceNumber, sequenceNumber)
+	m.endSequenceNumber = newEndSequenceNumber
+	m.arrivalTimes[m.index(sequenceNumber)] = arrivalTime
+}
+
+func (m *packetArrivalTimeMap) setNotReceived(startInclusive, endExclusive int64) {
+	for sn := startInclusive; sn < endExclusive; sn++ {
+		m.arrivalTimes[m.index(sn)] = -1
+	}
+}
+
+// BeginSequenceNumber returns the first valid sequence number in the map.
+func (m *packetArrivalTimeMap) BeginSequenceNumber() int64 {
+	return m.beginSequenceNumber
+}
+
+// EndSequenceNumber returns the first sequence number after the last valid sequence number in the map.
+func (m *packetArrivalTimeMap) EndSequenceNumber() int64 {
+	return m.endSequenceNumber
+}
+
+// FindNextAtOrAfter returns the sequence number and timestamp of the first received packet that has a sequence number
+// greator or equal to sequenceNumber.
+func (m *packetArrivalTimeMap) FindNextAtOrAfter(sequenceNumber int64) (foundSequenceNumber int64, arrivalTime int64, ok bool) {
+	for sequenceNumber = m.Clamp(sequenceNumber); sequenceNumber < m.endSequenceNumber; sequenceNumber++ {
+		if t := m.get(sequenceNumber); t >= 0 {
+			return sequenceNumber, t, true
+		}
+	}
+	return -1, -1, false
+}
+
+// EraseTo erases all elements from the beginning of the map until sequenceNumber.
+func (m *packetArrivalTimeMap) EraseTo(sequenceNumber int64) {
+	if sequenceNumber < m.beginSequenceNumber {
+		return
+	}
+	if sequenceNumber >= m.endSequenceNumber {
+		// Erase all.
+		m.beginSequenceNumber = m.endSequenceNumber
+		return
+	}
+	// Remove some
+	m.beginSequenceNumber = sequenceNumber
+	m.adjustToSize(int(m.endSequenceNumber - m.beginSequenceNumber))
+}
+
+// RemoveOldPackets removes packets from the beginning of the map as long as they are before
+// sequenceNumber and with an age older than arrivalTimeLimit.
+func (m *packetArrivalTimeMap) RemoveOldPackets(sequenceNumber int64, arrivalTimeLimit int64) {
+	checkTo := min64(sequenceNumber, m.endSequenceNumber)
+	for m.beginSequenceNumber < checkTo && m.get(m.beginSequenceNumber) <= arrivalTimeLimit {
+		m.beginSequenceNumber++
+	}
+	m.adjustToSize(int(m.endSequenceNumber - m.beginSequenceNumber))
+}
+
+// HasReceived returns whether a packet with the sequence number has been received.
+func (m *packetArrivalTimeMap) HasReceived(sequenceNumber int64) bool {
+	return m.get(sequenceNumber) >= 0
+}
+
+// Clamp returns sequenceNumber clamped to [beginSequenceNumber, endSequenceNumber]
+func (m *packetArrivalTimeMap) Clamp(sequenceNumber int64) int64 {
+	if sequenceNumber < m.beginSequenceNumber {
+		return m.beginSequenceNumber
+	}
+	if m.endSequenceNumber < sequenceNumber {
+		return m.endSequenceNumber
+	}
+	return sequenceNumber
+}
+
+func (m *packetArrivalTimeMap) get(sequenceNumber int64) int64 {
+	if sequenceNumber < m.beginSequenceNumber || sequenceNumber >= m.endSequenceNumber {
+		return -1
+	}
+	return m.arrivalTimes[m.index(sequenceNumber)]
+}
+
+func (m *packetArrivalTimeMap) index(sequenceNumber int64) int {
+	// Sequence number might be negative, and we always guarantee that arrivalTimes
+	// length is a power of 2, so it's easier to use "&" instead of "%"
+	return int(sequenceNumber & int64(m.capacity()-1))
+}
+
+func (m *packetArrivalTimeMap) adjustToSize(newSize int) {
+	if newSize > m.capacity() {
+		newCapacity := m.capacity()
+		for newCapacity < newSize {
+			newCapacity *= 2
+		}
+		m.reallocate(newCapacity)
+	}
+	if m.capacity() > max(minCapacity, newSize*4) {
+		newCapacity := m.capacity()
+		for newCapacity >= 2*max(newSize, minCapacity) {
+			newCapacity /= 2
+		}
+		m.reallocate(newCapacity)
+	}
+}
+
+func (m *packetArrivalTimeMap) capacity() int {
+	return len(m.arrivalTimes)
+}
+
+func (m *packetArrivalTimeMap) reallocate(newCapacity int) {
+	newBuffer := make([]int64, newCapacity)
+	for sn := m.beginSequenceNumber; sn < m.endSequenceNumber; sn++ {
+		newBuffer[int(sn&(int64(newCapacity-1)))] = m.get(sn)
+	}
+	m.arrivalTimes = newBuffer
+}
diff --git a/pkg/twcc/sender_interceptor.go b/pkg/twcc/sender_interceptor.go
index 8706e45..d7906fc 100644
--- a/pkg/twcc/sender_interceptor.go
+++ b/pkg/twcc/sender_interceptor.go
@@ -196,7 +196,7 @@ func (s *SenderInterceptor) loop(w interceptor.RTCPWriter) {
 		case <-ticker.C:
 			// build and send twcc
 			pkts := s.recorder.BuildFeedbackPacket()
-			if pkts == nil {
+			if len(pkts) == 0 {
 				continue
 			}
 			if _, err := w.Write(pkts, nil); err != nil {
diff --git a/pkg/twcc/twcc.go b/pkg/twcc/twcc.go
index f403998..c7246f3 100644
--- a/pkg/twcc/twcc.go
+++ b/pkg/twcc/twcc.go
@@ -6,95 +6,158 @@ package twcc
 
 import (
 	"math"
-	"sort"
 
+	"github.com/pion/interceptor/internal/sequencenumber"
 	"github.com/pion/rtcp"
 )
 
-type pktInfo struct {
-	sequenceNumber uint32
-	arrivalTime    int64
-}
+const (
+	packetWindowMicroseconds  = 500_000
+	maxMissingSequenceNumbers = 0x7FFE
+)
 
 // Recorder records incoming RTP packets and their delays and creates
 // transport wide congestion control feedback reports as specified in
 // https://datatracker.ietf.org/doc/html/draft-holmer-rmcat-transport-wide-cc-extensions-01
 type Recorder struct {
-	receivedPackets []pktInfo
+	arrivalTimeMap packetArrivalTimeMap
 
-	cycles             uint32
-	lastSequenceNumber uint16
+	sequenceUnwrapper sequencenumber.Unwrapper
+
+	// startSequenceNumber is the first sequence number that will be included in the the
+	// next feedback packet.
+	startSequenceNumber *int64
 
 	senderSSRC uint32
 	mediaSSRC  uint32
 	fbPktCnt   uint8
+
+	packetsHeld int
 }
 
 // NewRecorder creates a new Recorder which uses the given senderSSRC in the created
 // feedback packets.
 func NewRecorder(senderSSRC uint32) *Recorder {
 	return &Recorder{
-		receivedPackets: []pktInfo{},
-		senderSSRC:      senderSSRC,
+		senderSSRC: senderSSRC,
 	}
 }
 
 // Record marks a packet with mediaSSRC and a transport wide sequence number sequenceNumber as received at arrivalTime.
 func (r *Recorder) Record(mediaSSRC uint32, sequenceNumber uint16, arrivalTime int64) {
 	r.mediaSSRC = mediaSSRC
-	if sequenceNumber < 0x0fff && (r.lastSequenceNumber&0xffff) > 0xf000 {
-		r.cycles += 1 << 16
+
+	// "Unwrap" the sequence number to get a monotonically increasing sequence number that
+	// won't wrap around after math.MaxUint16.
+	unwrappedSN := r.sequenceUnwrapper.Unwrap(sequenceNumber)
+	r.maybeCullOldPackets(unwrappedSN, arrivalTime)
+	if r.startSequenceNumber == nil || unwrappedSN < *r.startSequenceNumber {
+		r.startSequenceNumber = &unwrappedSN
+	}
+
+	// We are only interested in the first time a packet is received.
+	if r.arrivalTimeMap.HasReceived(unwrappedSN) {
+		return
+	}
+
+	r.arrivalTimeMap.AddPacket(unwrappedSN, arrivalTime)
+	r.packetsHeld++
+
+	// Limit the range of sequence numbers to send feedback for.
+	if *r.startSequenceNumber < r.arrivalTimeMap.BeginSequenceNumber() {
+		sn := r.arrivalTimeMap.BeginSequenceNumber()
+		r.startSequenceNumber = &sn
+	}
+}
+
+func (r *Recorder) maybeCullOldPackets(sequenceNumber int64, arrivalTime int64) {
+	if r.startSequenceNumber != nil && *r.startSequenceNumber >= r.arrivalTimeMap.EndSequenceNumber() && arrivalTime >= packetWindowMicroseconds {
+		r.arrivalTimeMap.RemoveOldPackets(sequenceNumber, arrivalTime-packetWindowMicroseconds)
 	}
-	pkt := pktInfo{r.cycles | uint32(sequenceNumber), arrivalTime}
-	r.receivedPackets = append(r.receivedPackets, pkt)
-	r.lastSequenceNumber = sequenceNumber
 }
 
 // PacketsHeld returns the number of received packets currently held by the recorder
 func (r *Recorder) PacketsHeld() int {
-	return len(r.receivedPackets)
+	return r.packetsHeld
 }
 
 // BuildFeedbackPacket creates a new RTCP packet containing a TWCC feedback report.
 func (r *Recorder) BuildFeedbackPacket() []rtcp.Packet {
-	// sort received packets by sequence number, with earliest arrivalTime first in cases of duplicates
-	canBuild := false
-	sort.Slice(r.receivedPackets, func(i, j int) bool {
-		if r.receivedPackets[i].sequenceNumber == r.receivedPackets[j].sequenceNumber {
-			return r.receivedPackets[i].arrivalTime < r.receivedPackets[j].arrivalTime
-		}
-		canBuild = true // need at least 2 non-duplicate packets
-		return r.receivedPackets[i].sequenceNumber < r.receivedPackets[j].sequenceNumber
-	})
-	if !canBuild {
+	if r.startSequenceNumber == nil {
 		return nil
 	}
 
-	feedback := newFeedback(r.senderSSRC, r.mediaSSRC, r.fbPktCnt)
-	r.fbPktCnt++
-	feedback.setBase(
-		uint16(r.receivedPackets[0].sequenceNumber&0xffff),
-		r.receivedPackets[0].arrivalTime,
-	)
-
-	var pkts []rtcp.Packet
-	var prevSN uint32
-	for i, pkt := range r.receivedPackets {
-		if i == 0 || pkt.sequenceNumber != prevSN {
-			ok := feedback.addReceived(uint16(pkt.sequenceNumber&0xffff), pkt.arrivalTime)
-			if !ok {
-				pkts = append(pkts, feedback.getRTCP())
-				feedback = newFeedback(r.senderSSRC, r.mediaSSRC, r.fbPktCnt)
-				r.fbPktCnt++
-				feedback.addReceived(uint16(pkt.sequenceNumber&0xffff), pkt.arrivalTime)
+	endSN := r.arrivalTimeMap.EndSequenceNumber()
+	var feedbacks []rtcp.Packet
+	for *r.startSequenceNumber < endSN {
+		feedback := r.maybeBuildFeedbackPacket(*r.startSequenceNumber, endSN)
+		if feedback == nil {
+			break
+		}
+		feedbacks = append(feedbacks, feedback.getRTCP())
+
+		// NOTE: we don't erase packets from the history in case they need to be resent
+		// after a reordering. They will be removed instead in Record when they get too
+		// old.
+	}
+	r.packetsHeld = 0
+	return feedbacks
+}
+
+// maybeBuildFeedbackPacket builds a feedback packet starting from startSN (inclusive) until
+// endSN (exclusive).
+func (r *Recorder) maybeBuildFeedbackPacket(beginSeqNumInclusive, endSeqNumExclusive int64) *feedback {
+	// NOTE: The logic of this method is inspired by the implementation in Chrome.
+	// See https://source.chromium.org/chromium/chromium/src/+/refs/heads/main:third_party/webrtc/modules/remote_bitrate_estimator/remote_estimator_proxy.cc;l=276;drc=b5cd13bb6d5d157a5fbe3628b2dd1c1e106203c6
+	startSNInclusive, endSNExclusive := r.arrivalTimeMap.Clamp(beginSeqNumInclusive), r.arrivalTimeMap.Clamp(endSeqNumExclusive)
+
+	// Create feedback on demand, as we don't yet know if there are packets in the range that have been
+	// received.
+	var fb *feedback
+
+	nextSequenceNumber := beginSeqNumInclusive
+
+	for seq := startSNInclusive; seq < endSNExclusive; seq++ {
+		foundSeq, arrivalTime, ok := r.arrivalTimeMap.FindNextAtOrAfter(seq)
+		seq = foundSeq
+		if !ok || seq >= endSNExclusive {
+			break
+		}
+
+		if fb == nil {
+			fb = newFeedback(r.senderSSRC, r.mediaSSRC, r.fbPktCnt)
+			r.fbPktCnt++
+
+			// It should be possible to add seq to this new packet.
+			// If the difference between seq and beginSeqNumInclusive is too large, discard
+			// reporting too old missing packets.
+			baseSequenceNumber := max64(beginSeqNumInclusive, seq-maxMissingSequenceNumbers)
+
+			// baseSequenceNumber is the expected first sequence number. This is known,
+			// but we may not have actually received it, so the base time should be the time
+			// of the first received packet in the feedback.
+			fb.setBase(uint16(baseSequenceNumber), arrivalTime)
+
+			if !fb.addReceived(uint16(seq), arrivalTime) {
+				// Could not add a single received packet to the feedback.
+				// This is unexpected to actually occur, but if it does, we'll
+				// try again after skipping any missing packets.
+				// NOTE: It's fine that we already incremented fbPktCnt, as in essence
+				// we did actually "skip" a feedback (and this matches Chrome's behavior).
+				r.startSequenceNumber = &seq
+				return nil
 			}
+		} else if !fb.addReceived(uint16(seq), arrivalTime) {
+			// Could not add timestamp. Packet may be full. Return
+			// and try again with a fresh packet.
+			break
 		}
-		prevSN = pkt.sequenceNumber
+
+		nextSequenceNumber = seq + 1
 	}
-	r.receivedPackets = []pktInfo{}
-	pkts = append(pkts, feedback.getRTCP())
 
-	return pkts
+	r.startSequenceNumber = &nextSequenceNumber
+	return fb
 }
 
 type feedback struct {
@@ -274,9 +337,30 @@ func (c *chunk) reset() {
 	c.hasDifferentTypes = false
 }
 
+func max(a, b int) int {
+	if a > b {
+		return a
+	}
+	return b
+}
+
 func min(a, b int) int {
 	if a < b {
 		return a
 	}
 	return b
 }
+
+func max64(a, b int64) int64 {
+	if a > b {
+		return a
+	}
+	return b
+}
+
+func min64(a, b int64) int64 {
+	if a < b {
+		return a
+	}
+	return b
+}


Test patch:
diff --git a/pkg/twcc/arrival_time_map_test.go b/pkg/twcc/arrival_time_map_test.go
new file mode 100644
index 0000000..cf2103d
--- /dev/null
+++ b/pkg/twcc/arrival_time_map_test.go
@@ -0,0 +1,290 @@
+// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+package twcc
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestArrivalTimeMap(t *testing.T) {
+	t.Run("consistent when empty", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		assert.Equal(t, m.BeginSequenceNumber(), m.EndSequenceNumber())
+		assert.False(t, m.HasReceived(0))
+		assert.Equal(t, int64(0), m.Clamp(-5))
+		assert.Equal(t, int64(0), m.Clamp(5))
+	})
+
+	t.Run("inserts first item into map", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(42, 10)
+		assert.Equal(t, int64(42), m.BeginSequenceNumber())
+		assert.Equal(t, int64(43), m.EndSequenceNumber())
+
+		assert.False(t, m.HasReceived(41))
+		assert.True(t, m.HasReceived(42))
+		assert.False(t, m.HasReceived(43))
+		assert.False(t, m.HasReceived(44))
+
+		assert.Equal(t, int64(42), m.Clamp(-100))
+		assert.Equal(t, int64(42), m.Clamp(42))
+		assert.Equal(t, int64(43), m.Clamp(100))
+	})
+
+	t.Run("inserts with gaps", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(42, 0)
+		m.AddPacket(45, 11)
+		assert.Equal(t, int64(42), m.BeginSequenceNumber())
+		assert.Equal(t, int64(46), m.EndSequenceNumber())
+
+		assert.False(t, m.HasReceived(41))
+		assert.True(t, m.HasReceived(42))
+		assert.False(t, m.HasReceived(43))
+		assert.False(t, m.HasReceived(44))
+		assert.True(t, m.HasReceived(45))
+		assert.False(t, m.HasReceived(46))
+
+		assert.Equal(t, int64(0), m.get(42))
+		assert.Less(t, m.get(43), int64(0))
+		assert.Less(t, m.get(44), int64(0))
+		assert.Equal(t, int64(11), m.get(45))
+
+		assert.Equal(t, int64(42), m.Clamp(-100))
+		assert.Equal(t, int64(44), m.Clamp(44))
+		assert.Equal(t, int64(46), m.Clamp(100))
+	})
+
+	t.Run("find next at or after with gaps", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(42, 0)
+		m.AddPacket(45, 11)
+
+		seq, ts, ok := m.FindNextAtOrAfter(42)
+		assert.Equal(t, int64(42), seq)
+		assert.Equal(t, int64(0), ts)
+		assert.True(t, ok)
+
+		seq, ts, ok = m.FindNextAtOrAfter(43)
+		assert.Equal(t, int64(45), seq)
+		assert.Equal(t, int64(11), ts)
+		assert.True(t, ok)
+	})
+
+	t.Run("inserts within buffer", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(42, 10)
+		m.AddPacket(45, 11)
+
+		m.AddPacket(43, 12)
+		m.AddPacket(44, 13)
+
+		assert.False(t, m.HasReceived(41))
+		assert.True(t, m.HasReceived(42))
+		assert.True(t, m.HasReceived(43))
+		assert.True(t, m.HasReceived(44))
+		assert.True(t, m.HasReceived(45))
+		assert.False(t, m.HasReceived(46))
+
+		assert.Equal(t, int64(10), m.get(42))
+		assert.Equal(t, int64(12), m.get(43))
+		assert.Equal(t, int64(13), m.get(44))
+		assert.Equal(t, int64(11), m.get(45))
+	})
+
+	t.Run("grows buffer and removes old", func(t *testing.T) {
+		var m packetArrivalTimeMap
+
+		var largeSeqNum int64 = 42 + maxNumberOfPackets
+		m.AddPacket(42, 10)
+		m.AddPacket(43, 11)
+		m.AddPacket(44, 12)
+		m.AddPacket(45, 13)
+		m.AddPacket(largeSeqNum, 12)
+
+		assert.Equal(t, int64(43), m.BeginSequenceNumber())
+		assert.Equal(t, largeSeqNum+1, m.EndSequenceNumber())
+
+		assert.False(t, m.HasReceived(41))
+		assert.False(t, m.HasReceived(42))
+		assert.True(t, m.HasReceived(43))
+		assert.True(t, m.HasReceived(44))
+		assert.True(t, m.HasReceived(45))
+		assert.False(t, m.HasReceived(46))
+		assert.True(t, m.HasReceived(largeSeqNum))
+		assert.False(t, m.HasReceived(largeSeqNum+1))
+	})
+
+	t.Run("sequence number jump deletes all", func(t *testing.T) {
+		var m packetArrivalTimeMap
+
+		var largeSeqNum int64 = 42 + 2*maxNumberOfPackets
+		m.AddPacket(42, 10)
+		m.AddPacket(largeSeqNum, 12)
+
+		assert.Equal(t, largeSeqNum, m.BeginSequenceNumber())
+		assert.Equal(t, largeSeqNum+1, m.EndSequenceNumber())
+
+		assert.False(t, m.HasReceived(42))
+		assert.True(t, m.HasReceived(largeSeqNum))
+		assert.False(t, m.HasReceived(largeSeqNum+1))
+	})
+
+	t.Run("expands before beginning", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(42, 10)
+		m.AddPacket(-1000, 13)
+		assert.Equal(t, int64(-1000), m.BeginSequenceNumber())
+		assert.Equal(t, int64(43), m.EndSequenceNumber())
+
+		assert.False(t, m.HasReceived(-1001))
+		assert.True(t, m.HasReceived(-1000))
+		assert.False(t, m.HasReceived(-999))
+		assert.True(t, m.HasReceived(42))
+		assert.False(t, m.HasReceived(43))
+	})
+
+	t.Run("expanding before beginning keeps received", func(t *testing.T) {
+		var m packetArrivalTimeMap
+
+		var smallSeqNum int64 = 42 - 2*maxNumberOfPackets
+		m.AddPacket(42, 10)
+		m.AddPacket(smallSeqNum, 13)
+
+		assert.Equal(t, int64(42), m.BeginSequenceNumber())
+		assert.Equal(t, int64(43), m.EndSequenceNumber())
+	})
+
+	t.Run("erase to removes elements", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(42, 10)
+		m.AddPacket(43, 11)
+		m.AddPacket(44, 12)
+		m.AddPacket(45, 13)
+
+		m.EraseTo(44)
+
+		assert.Equal(t, int64(44), m.BeginSequenceNumber())
+		assert.Equal(t, int64(46), m.EndSequenceNumber())
+
+		assert.False(t, m.HasReceived(43))
+		assert.True(t, m.HasReceived(44))
+		assert.True(t, m.HasReceived(45))
+		assert.False(t, m.HasReceived(46))
+	})
+
+	t.Run("erases in empty map", func(t *testing.T) {
+		var m packetArrivalTimeMap
+
+		assert.Equal(t, m.BeginSequenceNumber(), m.EndSequenceNumber())
+
+		m.EraseTo(m.EndSequenceNumber())
+		assert.Equal(t, m.BeginSequenceNumber(), m.EndSequenceNumber())
+	})
+
+	t.Run("is tolerant to wrong arguments for erase", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(42, 10)
+		m.AddPacket(43, 11)
+
+		m.EraseTo(1)
+
+		assert.Equal(t, int64(42), m.BeginSequenceNumber())
+		assert.Equal(t, int64(44), m.EndSequenceNumber())
+
+		m.EraseTo(100)
+
+		assert.Equal(t, int64(44), m.BeginSequenceNumber())
+		assert.Equal(t, int64(44), m.EndSequenceNumber())
+	})
+
+	//nolint:dupl
+	t.Run("erase all remembers beginning sequence number", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(42, 10)
+		m.AddPacket(43, 11)
+		m.AddPacket(44, 12)
+		m.AddPacket(45, 13)
+
+		m.EraseTo(46)
+		m.AddPacket(50, 10)
+
+		assert.Equal(t, int64(46), m.BeginSequenceNumber())
+		assert.Equal(t, int64(51), m.EndSequenceNumber())
+
+		assert.False(t, m.HasReceived(45))
+		assert.False(t, m.HasReceived(46))
+		assert.False(t, m.HasReceived(47))
+		assert.False(t, m.HasReceived(48))
+		assert.False(t, m.HasReceived(49))
+		assert.True(t, m.HasReceived(50))
+		assert.False(t, m.HasReceived(51))
+	})
+
+	//nolint:dupl
+	t.Run("erase to missing sequence number", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(37, 10)
+		m.AddPacket(39, 11)
+		m.AddPacket(40, 12)
+		m.AddPacket(41, 13)
+
+		m.EraseTo(38)
+
+		m.AddPacket(42, 40)
+
+		assert.Equal(t, int64(38), m.BeginSequenceNumber())
+		assert.Equal(t, int64(43), m.EndSequenceNumber())
+
+		assert.False(t, m.HasReceived(37))
+		assert.False(t, m.HasReceived(38))
+		assert.True(t, m.HasReceived(39))
+		assert.True(t, m.HasReceived(40))
+		assert.True(t, m.HasReceived(41))
+		assert.True(t, m.HasReceived(42))
+		assert.False(t, m.HasReceived(43))
+	})
+
+	t.Run("remove old packets", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(37, 10)
+		m.AddPacket(39, 11)
+		m.AddPacket(40, 12)
+		m.AddPacket(41, 13)
+
+		m.RemoveOldPackets(42, 11)
+
+		assert.Equal(t, int64(40), m.BeginSequenceNumber())
+		assert.Equal(t, int64(42), m.EndSequenceNumber())
+
+		assert.False(t, m.HasReceived(39))
+		assert.True(t, m.HasReceived(40))
+		assert.True(t, m.HasReceived(41))
+		assert.False(t, m.HasReceived(42))
+	})
+
+	t.Run("shrinks buffer when necessary", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		var largeSeqNum int64 = 100 + maxNumberOfPackets - 1
+		m.AddPacket(100, 10)
+		m.AddPacket(largeSeqNum, 11)
+
+		m.EraseTo(largeSeqNum - 1)
+
+		assert.Equal(t, largeSeqNum-1, m.BeginSequenceNumber())
+		assert.Equal(t, largeSeqNum+1, m.EndSequenceNumber())
+
+		assert.Equal(t, minCapacity, m.capacity())
+	})
+
+	t.Run("find next at or after with invalid sequence", func(t *testing.T) {
+		var m packetArrivalTimeMap
+		m.AddPacket(100, 10)
+
+		_, _, ok := m.FindNextAtOrAfter(101)
+		assert.False(t, ok)
+	})
+}
diff --git a/pkg/twcc/twcc_test.go b/pkg/twcc/twcc_test.go
index c829159..7e41352 100644
--- a/pkg/twcc/twcc_test.go
+++ b/pkg/twcc/twcc_test.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/pion/rtcp"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func rtcpToTwcc(t *testing.T, in []rtcp.Packet) []*rtcp.TransportLayerCC {
@@ -419,8 +420,12 @@ func increaseTime(arrivalTime *int64, increaseAmount int64) int64 {
 
 func marshalAll(t *testing.T, pkts []rtcp.Packet) {
 	for _, pkt := range pkts {
-		_, err := pkt.Marshal()
+		marshaled, err := pkt.Marshal()
 		assert.NoError(t, err)
+
+		// Chrome expects feedback packets to always be 18 bytes or more.
+		// https://source.chromium.org/chromium/chromium/src/+/main:third_party/webrtc/modules/rtp_rtcp/source/rtcp_packet/transport_feedback.cc;l=423?q=transport_feedback.cc&ss=chromium%2Fchromium%2Fsrc
+		assert.GreaterOrEqual(t, len(marshaled), 18)
 	}
 }
 
@@ -493,7 +498,7 @@ func TestBuildFeedbackPacket_Rolling(t *testing.T) {
 	rtcpPackets := r.BuildFeedbackPacket()
 	assert.Equal(t, 1, len(rtcpPackets))
 
-	addRun(t, r, []uint16{4, 8, 9, 10}, []int64{
+	addRun(t, r, []uint16{0, 4, 5, 6}, []int64{
 		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
 		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
 		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
@@ -512,7 +517,7 @@ func TestBuildFeedbackPacket_Rolling(t *testing.T) {
 		},
 		SenderSSRC:         5000,
 		MediaSSRC:          5000,
-		BaseSequenceNumber: 4,
+		BaseSequenceNumber: 0,
 		ReferenceTime:      1,
 		FbPktCount:         1,
 		PacketStatusCount:  7,
@@ -550,42 +555,78 @@ func TestBuildFeedbackPacket_MinInput(t *testing.T) {
 	})
 
 	pkts := r.BuildFeedbackPacket()
-	assert.Nil(t, pkts)
-
-	addRun(t, r, []uint16{1}, []int64{
-		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
-	})
-
-	pkts = r.BuildFeedbackPacket()
 	assert.Equal(t, 1, len(pkts))
 
 	assert.Equal(t, &rtcp.TransportLayerCC{
 		Header: rtcp.Header{
-			Count:  rtcp.FormatTCC,
-			Type:   rtcp.TypeTransportSpecificFeedback,
-			Length: 5,
+			Count:   rtcp.FormatTCC,
+			Type:    rtcp.TypeTransportSpecificFeedback,
+			Length:  5,
+			Padding: true,
 		},
 		SenderSSRC:         5000,
 		MediaSSRC:          5000,
 		BaseSequenceNumber: 0,
 		ReferenceTime:      1,
 		FbPktCount:         0,
-		PacketStatusCount:  2,
+		PacketStatusCount:  1,
 		PacketChunks: []rtcp.PacketStatusChunk{
 			&rtcp.RunLengthChunk{
 				PacketStatusSymbol: 1,
 				Type:               rtcp.TypeTCCRunLengthChunk,
-				RunLength:          2,
+				RunLength:          1,
 			},
 		},
 		RecvDeltas: []*rtcp.RecvDelta{
 			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 0},
-			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: rtcp.TypeTCCDeltaScaleFactor},
 		},
 	}, rtcpToTwcc(t, pkts)[0])
 	marshalAll(t, pkts)
 }
 
+func TestBuildFeedbackPacket_MissingPacketsBetweenFeedbacks(t *testing.T) {
+	r := NewRecorder(5000)
+
+	// Create a run of received packets.
+	arrivalTime := int64(scaleFactorReferenceTime)
+	addRun(t, r, []uint16{0, 1, 2, 3}, []int64{
+		scaleFactorReferenceTime,
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+	})
+	rtcpPackets := r.BuildFeedbackPacket()
+	assert.Equal(t, 1, len(rtcpPackets))
+
+	// Now create another run of received packets, but with a gap.
+	addRun(t, r, []uint16{7, 8, 9}, []int64{
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor*256),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+	})
+	rtcpPackets = r.BuildFeedbackPacket()
+	require.Equal(t, 1, len(rtcpPackets))
+	twccPacket := rtcpToTwcc(t, rtcpPackets)[0]
+	assert.Equal(t, uint16(4), twccPacket.BaseSequenceNumber, "Base sequence should be one after the end of the previous feedback")
+	assert.Equal(t, uint16(6), twccPacket.PacketStatusCount, "Feedback should include status for both the lost and received packets")
+	expectedPacketChunks := []rtcp.PacketStatusChunk{
+		&rtcp.StatusVectorChunk{
+			Type:       rtcp.TypeTCCRunLengthChunk,
+			SymbolSize: rtcp.TypeTCCSymbolSizeTwoBit,
+			SymbolList: []uint16{
+				rtcp.TypeTCCPacketNotReceived,
+				rtcp.TypeTCCPacketNotReceived,
+				rtcp.TypeTCCPacketNotReceived,
+				rtcp.TypeTCCPacketReceivedSmallDelta,
+				rtcp.TypeTCCPacketReceivedSmallDelta,
+				rtcp.TypeTCCPacketReceivedSmallDelta,
+			},
+		},
+	}
+	assert.Equal(t, expectedPacketChunks, twccPacket.PacketChunks)
+	marshalAll(t, rtcpPackets)
+}
+
 func TestBuildFeedbackPacketCount(t *testing.T) {
 	r := NewRecorder(5000)
 
@@ -618,10 +659,10 @@ func TestDuplicatePackets(t *testing.T) {
 
 	arrivalTime := int64(scaleFactorReferenceTime)
 	addRun(t, r, []uint16{12, 13, 13, 14}, []int64{
-		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
-		arrivalTime + rtcp.TypeTCCDeltaScaleFactor*256,
-		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
-		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		arrivalTime,
+		arrivalTime,
+		arrivalTime,
+		arrivalTime,
 	})
 
 	rtcpPackets := r.BuildFeedbackPacket()
@@ -649,9 +690,9 @@ func TestDuplicatePackets(t *testing.T) {
 			},
 		},
 		RecvDeltas: []*rtcp.RecvDelta{
-			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 250},
-			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 250},
-			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 250},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 0},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 0},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 0},
 		},
 	}, rtcpToTwcc(t, rtcpPackets)[0])
 	marshalAll(t, rtcpPackets)
@@ -848,7 +889,7 @@ func TestReorderedPackets(t *testing.T) {
 	marshalAll(t, rtcpPackets)
 }
 
-func TestPacketsHeld(t *testing.T) {
+func TestPacketsHheld(t *testing.T) {
 	r := NewRecorder(5000)
 	assert.Zero(t, r.PacketsHeld())