From 4a9b84478188277cf186a855b60ebf9b7d97519a Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Sat, 5 Feb 2022 23:25:53 +0530
Subject: [PATCH] Process RTX packets for TWCC (#406)

The SRTP replay detection was disabled recently.
But, they were effectively getting dropped in `sfu.bucket`.

Doing two things with RTX packets in this PR
1. Update stats - add to packet count and bytes
2. Process header extension - to process TWCC
---
 pkg/sfu/buffer/bucket.go |  4 ++--
 pkg/sfu/buffer/buffer.go | 48 +++++++++++++++++++++++++++++-----------
 2 files changed, 37 insertions(+), 15 deletions(-)

diff --git a/pkg/sfu/buffer/bucket.go b/pkg/sfu/buffer/bucket.go
index f853de218..9c9dee3dc 100644
--- a/pkg/sfu/buffer/bucket.go
+++ b/pkg/sfu/buffer/bucket.go
@@ -33,8 +33,8 @@ func (b *Bucket) AddPacket(pkt []byte) ([]byte, error) {
 		b.init = true
 	}
 	diff := sn - b.headSN
-	if diff > (1 << 15) {
-		// out-of-order
+	if diff == 0 || diff > (1<<15) {
+		// duplicate of lsat packet or out-of-order
 		return b.set(sn, pkt)
 	}
 	b.headSN = sn
diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go
index 041e52f62..145f38af7 100644
--- a/pkg/sfu/buffer/buffer.go
+++ b/pkg/sfu/buffer/buffer.go
@@ -272,24 +272,44 @@ func (b *Buffer) OnClose(fn func()) {
 }
 
 func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
+	isRTX := false
+
 	pb, err := b.bucket.AddPacket(pkt)
 	if err != nil {
 		if err != ErrRTXPacket {
 			b.logger.Warnw("could not add RTP packet to bucket", err)
+			return
+		} else {
+			isRTX = true
 		}
-		return
 	}
 
 	var p rtp.Packet
-	if err := p.Unmarshal(pb); err != nil {
+	if isRTX {
+		err = p.Unmarshal(pkt)
+	} else {
+		err = p.Unmarshal(pb)
+	}
+	if err != nil {
 		b.logger.Warnw("error unmarshaling RTP packet", err)
 		return
 	}
 
-	b.updateStreamState(&p, len(pkt), arrivalTime)
+	b.updateStreamState(&p, len(pkt), arrivalTime, isRTX)
 
 	b.processHeaderExtensions(&p, arrivalTime)
 
+	if isRTX {
+		//
+		// Run RTX packets through
+		//  1. state update - to update stats
+		//  2. TWCC just in case remote side is retransmitting an old packet for probing
+		//
+		// But, do not forward those packets
+		//
+		return
+	}
+
 	ep, temporalLayer := b.getExtPacket(pb, &p, arrivalTime)
 	if ep == nil {
 		return
@@ -305,7 +325,7 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
 	b.doReports(arrivalTime)
 }
 
-func (b *Buffer) updateStreamState(p *rtp.Packet, pktSize int, arrivalTime int64) {
+func (b *Buffer) updateStreamState(p *rtp.Packet, pktSize int, arrivalTime int64, isRTX bool) {
 	sn := p.SequenceNumber
 
 	if b.stats.PacketCount == 0 {
@@ -344,17 +364,19 @@ func (b *Buffer) updateStreamState(p *rtp.Packet, pktSize int, arrivalTime int64
 	b.stats.PacketCount++
 	b.stats.TotalBytes += uint64(pktSize)
 
-	// jitter
-	arrival := uint32(arrivalTime / 1e6 * int64(b.clockRate/1e3))
-	transit := arrival - p.Timestamp
-	if b.lastTransit != 0 {
-		d := int32(transit - b.lastTransit)
-		if d < 0 {
-			d = -d
+	if !isRTX {
+		// jitter
+		arrival := uint32(arrivalTime / 1e6 * int64(b.clockRate/1e3))
+		transit := arrival - p.Timestamp
+		if b.lastTransit != 0 {
+			d := int32(transit - b.lastTransit)
+			if d < 0 {
+				d = -d
+			}
+			b.stats.Jitter += (float64(d) - b.stats.Jitter) / 16
 		}
-		b.stats.Jitter += (float64(d) - b.stats.Jitter) / 16
+		b.lastTransit = transit
 	}
-	b.lastTransit = transit
 }
 
 func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) {