Repository: pion/interceptor
Pull request #197 (2023-07-26 18:42:23)

Don't use out-of-order packets to calculate RTPTime for sender reports
#### Description

If a sender report is generated immediately after sending an out-of-order packet (which can happen e.g. in an SFU when forwarding media), the timestamp from the last in-order packet should be extrapolated, since the departure time of an out-of-order packet is is not properly correlated with its RTP timestamp.

Leave an option to re-enable the old behavior.



codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pion/interceptor/pull/197?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
Patch coverage: **`100.00%`** and project coverage change: **`-0.19%`** :warning:
> 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 [(`c7f8c17`)](https://app.codecov.io/gh/pion/interceptor/pull/197?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) 79.22%.

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


```diff
@@            Coverage Diff             @@
##           master     #197      +/-   ##
==========================================
- Coverage   79.41%   79.22%   -0.19%     
==========================================
  Files          67       67              
  Lines        3367     3375       +8     
==========================================
  Hits         2674     2674              
- Misses        574      580       +6     
- Partials      119      121       +2     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `79.22% <100.00%> (-0.19%)` | :arrow_down: |
| wasm | `77.24% <100.00%> (+0.05%)` | :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/197?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [pkg/report/sender\_interceptor.go](https://app.codecov.io/gh/pion/interceptor/pull/197?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3JlcG9ydC9zZW5kZXJfaW50ZXJjZXB0b3IuZ28=) | `82.19% <100.00%> (ø)` | |
| [pkg/report/sender\_option.go](https://app.codecov.io/gh/pion/interceptor/pull/197?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3JlcG9ydC9zZW5kZXJfb3B0aW9uLmdv) | `100.00% <100.00%> (ø)` | |
| [pkg/report/sender\_stream.go](https://app.codecov.io/gh/pion/interceptor/pull/197?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3JlcG9ydC9zZW5kZXJfc3RyZWFtLmdv) | `100.00% <100.00%> (ø)` | |

... and [2 files with indirect coverage changes](https://app.codecov.io/gh/pion/interceptor/pull/197/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/197?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: The motivation behind this PR is that we were getting reports of A/V desync from users of our SFU. After some investigation, it seemed like the issue was related to scenarios where any reordering of packets occurred (such as RTX after loss). Depending on how severe the reordering is, it can be the case where the most recently sent packet is actually quite old. This can cause problems with Chrome's "[RTP to NTP estimator](https://source.chromium.org/chromium/chromium/src/+/refs/heads/main:third_party/webrtc/system_wrappers/source/rtp_to_ntp_estimator.cc;l=75;drc=b5cd13bb6d5d157a5fbe3628b2dd1c1e106203c6;bpv=1;bpt=1)". This estimator works by looking at the most recent 20 Sender Reports and running a linear regression to correlate RTP timestamps to NTP timestamps (which are supposed to represent the same instant in time for the values in Sender Reports).

The RTP to NTP estimator (along with a clock offset estimator) is then used to calculate playout timestamps for the respective audio and video tracks in local time by converting the RTP timestamps for the most recently played media frames; if the playout timestamps differ, then either the audio or video jitter buffer may be grown or shrunk to try to keep them in sync.

Before this PR, it was possible for an old packet to be the most recently arrived/sent packet, so the RTP timestamp would be skewed to look older than it should. This would, in turn, impact the RTP estimator for up to 20 seconds (until the bad SR is cycled out), skewing the playout timestamp estimator, and ultimately actually causing A/V desync when Chrome tries to correct desync that doesn't actually exist.
aler9: Hello,

The RTP-to-NTP analysis you performed is right.

The issue you encountered is related to the special case in which a SFU deliberately routes unordered packets from an origin to clients without performing reordering. While this does minimize latency, it also causes clients to concurrently emit NACKs whenever there are issues in the origin-to-server connection, not in the server-to-client connections, potentially exhausting the SFU bandwidth when there are thousands of clients.

If you instead perform packet reordering and NACK handling on the SFU and recover packets directly on the SFU, clients won't detect losses (unless packets can't be recovered in the origin-to-server connection), won't emit NACKs and won't have NTP/RTP issues.

Therefore this PR is useless if a SFU is implemented in the right way and if the sender interceptor is placed after the reordering mechanism.
adriancable: @cptpcrd - I support this PR. Can you kindly rebase it on the base branch and then I can merge it.
kcaffrey: This is now rebased @adriancable. Thanks!

Patch:
diff --git a/pkg/report/sender_interceptor.go b/pkg/report/sender_interceptor.go
index 508d7e9..087838e 100644
--- a/pkg/report/sender_interceptor.go
+++ b/pkg/report/sender_interceptor.go
@@ -59,6 +59,8 @@ type SenderInterceptor struct {
 	wg        sync.WaitGroup
 	close     chan struct{}
 	started   chan struct{}
+
+	useLatestPacket bool
 }
 
 func (s *SenderInterceptor) isClosed() bool {
@@ -133,7 +135,7 @@ func (s *SenderInterceptor) loop(rtcpWriter interceptor.RTCPWriter) {
 // BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method
 // will be called once per rtp packet.
 func (s *SenderInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
-	stream := newSenderStream(info.SSRC, info.ClockRate)
+	stream := newSenderStream(info.SSRC, info.ClockRate, s.useLatestPacket)
 	s.streams.Store(info.SSRC, stream)
 
 	return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, a interceptor.Attributes) (int, error) {
diff --git a/pkg/report/sender_option.go b/pkg/report/sender_option.go
index 50abafb..1e489b0 100644
--- a/pkg/report/sender_option.go
+++ b/pkg/report/sender_option.go
@@ -44,6 +44,15 @@ func SenderTicker(f TickerFactory) SenderOption {
 	}
 }
 
+// SenderUseLatestPacket sets the interceptor to always use the latest packet, even
+// if it appears to be out-of-order.
+func SenderUseLatestPacket() SenderOption {
+	return func(r *SenderInterceptor) error {
+		r.useLatestPacket = true
+		return nil
+	}
+}
+
 // enableStartTracking is used by tests to synchronize whether the loop() has begun
 // and it's safe to start sending ticks to the ticker.
 func enableStartTracking(startedCh chan struct{}) SenderOption {
diff --git a/pkg/report/sender_stream.go b/pkg/report/sender_stream.go
index 29d14da..6c708bd 100644
--- a/pkg/report/sender_stream.go
+++ b/pkg/report/sender_stream.go
@@ -17,17 +17,21 @@ type senderStream struct {
 	clockRate float64
 	m         sync.Mutex
 
+	useLatestPacket bool
+
 	// data from rtp packets
 	lastRTPTimeRTP  uint32
 	lastRTPTimeTime time.Time
+	lastRTPSN       uint16
 	packetCount     uint32
 	octetCount      uint32
 }
 
-func newSenderStream(ssrc uint32, clockRate uint32) *senderStream {
+func newSenderStream(ssrc uint32, clockRate uint32, useLatestPacket bool) *senderStream {
 	return &senderStream{
-		ssrc:      ssrc,
-		clockRate: float64(clockRate),
+		ssrc:            ssrc,
+		clockRate:       float64(clockRate),
+		useLatestPacket: useLatestPacket,
 	}
 }
 
@@ -35,9 +39,12 @@ func (stream *senderStream) processRTP(now time.Time, header *rtp.Header, payloa
 	stream.m.Lock()
 	defer stream.m.Unlock()
 
-	// always update time to minimize errors
-	stream.lastRTPTimeRTP = header.Timestamp
-	stream.lastRTPTimeTime = now
+	if stream.useLatestPacket || stream.packetCount == 0 || int16(header.SequenceNumber-stream.lastRTPSN) > 0 {
+		// Told to consider every packet, or this was the first packet, or it's in-order
+		stream.lastRTPSN = header.SequenceNumber
+		stream.lastRTPTimeRTP = header.Timestamp
+		stream.lastRTPTimeTime = now
+	}
 
 	stream.packetCount++
 	stream.octetCount += uint32(len(payload))


Test patch:
diff --git a/pkg/report/sender_interceptor_test.go b/pkg/report/sender_interceptor_test.go
index cc034fb..930be6d 100644
--- a/pkg/report/sender_interceptor_test.go
+++ b/pkg/report/sender_interceptor_test.go
@@ -92,6 +92,130 @@ func TestSenderInterceptor(t *testing.T) {
 		}, sr)
 	})
 
+	t.Run("out of order RTP packets", func(t *testing.T) {
+		mt := &test.MockTime{}
+		f, err := NewSenderInterceptor(
+			SenderInterval(time.Millisecond*50),
+			SenderLog(logging.NewDefaultLoggerFactory().NewLogger("test")),
+			SenderNow(mt.Now),
+		)
+		assert.NoError(t, err)
+
+		i, err := f.NewInterceptor("")
+		assert.NoError(t, err)
+
+		stream := test.NewMockStream(&interceptor.StreamInfo{
+			SSRC:      123456,
+			ClockRate: 90000,
+		}, i)
+		defer func() {
+			assert.NoError(t, stream.Close())
+		}()
+
+		// Write several packets
+		for i := 0; i < 10; i++ {
+			assert.NoError(t, stream.WriteRTP(&rtp.Packet{
+				Header: rtp.Header{
+					SequenceNumber: uint16(i),
+					Timestamp:      uint32(i),
+				},
+				Payload: []byte("\x00\x00"),
+			}))
+		}
+
+		// Skip a packet, then redeliver it out-of-order
+		assert.NoError(t, stream.WriteRTP(&rtp.Packet{
+			Header: rtp.Header{
+				SequenceNumber: 12,
+				Timestamp:      12,
+			},
+			Payload: []byte("\x00\x00"),
+		}))
+		assert.NoError(t, stream.WriteRTP(&rtp.Packet{
+			Header: rtp.Header{
+				SequenceNumber: 11,
+				Timestamp:      11,
+			},
+			Payload: []byte("\x00\x00"),
+		}))
+
+		pkts := <-stream.WrittenRTCP()
+		assert.Equal(t, len(pkts), 1)
+		sr, ok := pkts[0].(*rtcp.SenderReport)
+		assert.True(t, ok)
+		// The out-of-order packet is included in PacketCount and OctetCount, but the RTP
+		// timestamp of the last in-order packet is used for RTPTime
+		assert.Equal(t, &rtcp.SenderReport{
+			SSRC:        123456,
+			NTPTime:     ntp.ToNTP(mt.Now()),
+			RTPTime:     12,
+			PacketCount: 12,
+			OctetCount:  24,
+		}, sr)
+	})
+
+	t.Run("out of order RTP packets with SenderUseLatestPacket", func(t *testing.T) {
+		mt := &test.MockTime{}
+		f, err := NewSenderInterceptor(
+			SenderInterval(time.Millisecond*50),
+			SenderLog(logging.NewDefaultLoggerFactory().NewLogger("test")),
+			SenderNow(mt.Now),
+			SenderUseLatestPacket(),
+		)
+		assert.NoError(t, err)
+
+		i, err := f.NewInterceptor("")
+		assert.NoError(t, err)
+
+		stream := test.NewMockStream(&interceptor.StreamInfo{
+			SSRC:      123456,
+			ClockRate: 90000,
+		}, i)
+		defer func() {
+			assert.NoError(t, stream.Close())
+		}()
+
+		// Write several packets
+		for i := 0; i < 10; i++ {
+			assert.NoError(t, stream.WriteRTP(&rtp.Packet{
+				Header: rtp.Header{
+					SequenceNumber: uint16(i),
+					Timestamp:      uint32(i),
+				},
+				Payload: []byte("\x00\x00"),
+			}))
+		}
+
+		// Skip a packet, then redeliver it out-of-order
+		assert.NoError(t, stream.WriteRTP(&rtp.Packet{
+			Header: rtp.Header{
+				SequenceNumber: 12,
+				Timestamp:      12,
+			},
+			Payload: []byte("\x00\x00"),
+		}))
+		assert.NoError(t, stream.WriteRTP(&rtp.Packet{
+			Header: rtp.Header{
+				SequenceNumber: 11,
+				Timestamp:      11,
+			},
+			Payload: []byte("\x00\x00"),
+		}))
+
+		pkts := <-stream.WrittenRTCP()
+		assert.Equal(t, len(pkts), 1)
+		sr, ok := pkts[0].(*rtcp.SenderReport)
+		assert.True(t, ok)
+		// The out-of-order packet *is*  used for RTPTime
+		assert.Equal(t, &rtcp.SenderReport{
+			SSRC:        123456,
+			NTPTime:     ntp.ToNTP(mt.Now()),
+			RTPTime:     11,
+			PacketCount: 12,
+			OctetCount:  24,
+		}, sr)
+	})
+
 	t.Run("inject ticker", func(t *testing.T) {
 		mNow := &test.MockTime{}
 		mTick := &test.MockTicker{