Don't drop message if calculate duration is too small (#3442)

* Don't drop message if calculate duration is too small

* fix test

go.mod

@@ -36,7 +36,7 @@ require (
 	github.com/pion/interceptor v0.1.37
 	github.com/pion/rtcp v1.2.15
 	github.com/pion/rtp v1.8.11
-	github.com/pion/sctp v1.8.35
+	github.com/pion/sctp v1.8.36
 	github.com/pion/sdp/v3 v3.0.10
 	github.com/pion/transport/v3 v3.0.7
 	github.com/pion/turn/v4 v4.0.0

go.sum

@@ -251,8 +251,8 @@ github.com/pion/rtcp v1.2.15 h1:LZQi2JbdipLOj4eBjK4wlVoQWfrZbh3Q6eHtWtJBZBo=
 github.com/pion/rtcp v1.2.15/go.mod h1:jlGuAjHMEXwMUHK78RgX0UmEJFV4zUKOFHR7OP+D3D0=
 github.com/pion/rtp v1.8.11 h1:17xjnY5WO5hgO6SD3/NTIUPvSFw/PbLsIJyz1r1yNIk=
 github.com/pion/rtp v1.8.11/go.mod h1:8uMBJj32Pa1wwx8Fuv/AsFhn8jsgw+3rUC2PfoBZ8p4=
-github.com/pion/sctp v1.8.35 h1:qwtKvNK1Wc5tHMIYgTDJhfZk7vATGVHhXbUDfHbYwzA=
-github.com/pion/sctp v1.8.35/go.mod h1:EcXP8zCYVTRy3W9xtOF7wJm1L1aXfKRQzaM33SjQlzg=
+github.com/pion/sctp v1.8.36 h1:owNudmnz1xmhfYje5L/FCav3V9wpPRePHle3Zi+P+M0=
+github.com/pion/sctp v1.8.36/go.mod h1:cNiLdchXra8fHQwmIoqw0MbLLMs+f7uQ+dGMG2gWebE=
 github.com/pion/sdp/v3 v3.0.10 h1:6MChLE/1xYB+CjumMw+gZ9ufp2DPApuVSnDT8t5MIgA=
 github.com/pion/sdp/v3 v3.0.10/go.mod h1:88GMahN5xnScv1hIMTqLdu/cOcUkj6a9ytbncwMCq2E=
 github.com/pion/srtp/v3 v3.0.4 h1:2Z6vDVxzrX3UHEgrUyIGM4rRouoC7v+NiF1IHtp9B5M=

pkg/sfu/datachannel/bitrate.go

@@ -50,6 +50,10 @@ func (c *BitrateCalculator) AddBytes(bytes int, bufferedAmout int, ts time.Time)
 	defer c.lock.Unlock()
 
 	bytes -= bufferedAmout - c.lastBufferedAmount
+	if bytes < 0 {
+		// it is possible that internal buffering (non-data like DCEP packet from webrtc) caused bytes to be negative
+		bytes = 0
+	}
 	c.lastBufferedAmount = bufferedAmout
 	if ts.Sub(c.active.start) >= c.windowDuration {
 		c.windows.PushBack(c.active)
@@ -76,15 +80,15 @@ func (c *BitrateCalculator) AddBytes(bytes int, bufferedAmout int, ts time.Time)
 
 }
 
-func (c *BitrateCalculator) Bitrate(ts time.Time) int {
+func (c *BitrateCalculator) Bitrate(ts time.Time) (int, bool) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	duration := ts.Sub(c.start)
 	if duration < c.windowDuration {
-		duration = c.windowDuration
+		return 0, false
 	}
 
-	return c.bytes * 8 * 1000 / int(duration.Milliseconds())
+	return c.bytes * 8 * 1000 / int(duration.Milliseconds()), true
 }
 
 type bitrateWindow struct {

pkg/sfu/datachannel/bitrate_test.go

@@ -15,15 +15,22 @@ func TestBitrateCalculator(t *testing.T) {
 	c.AddBytes(100, 0, t0)
 	// bytes buffered
 	c.AddBytes(100, 100, t0.Add(50*time.Millisecond))
+	bitrate, ok := c.Bitrate(t0.Add(50 * time.Millisecond))
+	require.Equal(t, 0, bitrate)
+	require.False(t, ok)
 	// 50 bytes sent (50 bytes buffer flushed)
 	c.AddBytes(100, 50, t0.Add(time.Second))
 
 	// 250 bytes sent in 1 second
-	require.Equal(t, 2000, c.Bitrate(t0.Add(time.Second)))
+	bitrate, ok = c.Bitrate(t0.Add(time.Second))
+	require.Equal(t, 2000, bitrate)
+	require.True(t, ok)
 
 	// silence for long time
 	t1 := t0.Add(2 * BitrateDuration)
 	// 150 bytes sent (50 bytes buffer flushed)
 	c.AddBytes(100, 0, t1)
-	require.Equal(t, 1200, c.Bitrate(t1.Add(time.Second)))
+	bitrate, ok = c.Bitrate(t1.Add(time.Second))
+	require.Equal(t, 1200, bitrate)
+	require.True(t, ok)
 }

pkg/sfu/datachannel/datachannel_writer.go

@@ -64,7 +64,7 @@ func (w *DataChannelWriter[T]) Write(p []byte) (n int, err error) {
 		w.rate.AddBytes(n, int(w.bufferGetter.BufferedAmount()), now)
 		// retry if the write timed out on a non-slow receiver
 		if errors.Is(err, context.DeadlineExceeded) {
-			if bitrate := w.rate.Bitrate(now); bitrate >= w.slowThreshold {
+			if bitrate, ok := w.rate.Bitrate(now); !ok || bitrate >= w.slowThreshold {
 				continue
 			} else {
 				err = fmt.Errorf("%w: bitrate %d, threshold %d", ErrDataDroppedBySlowReader, bitrate, w.slowThreshold)

test/client/datachannel_reader.go

@@ -20,7 +20,7 @@ func NewDataChannelReader(bitrate int) *DataChannelReader {
 
 func (d *DataChannelReader) Read(p []byte, sid string) {
 	for {
-		if bitrate := d.bitrate.Bitrate(time.Now()); bitrate > 0 && bitrate > d.target {
+		if bitrate, ok := d.bitrate.Bitrate(time.Now()); ok && bitrate > 0 && bitrate > d.target {
 			time.Sleep(10 * time.Millisecond)
 			d.bitrate.AddBytes(0, 0, time.Now())
 			continue
@@ -31,5 +31,6 @@ func (d *DataChannelReader) Read(p []byte, sid string) {
 }
 
 func (d *DataChannelReader) Bitrate() int {
-	return d.bitrate.Bitrate(time.Now())
+	b, _ := d.bitrate.Bitrate(time.Now())
+	return b
 }