From a1dc7bcfd7ef8d5a604d57f0e627874be8a8bf3f Mon Sep 17 00:00:00 2001 From: Mathis Engelbart Date: Tue, 10 Mar 2026 15:14:50 +0100 Subject: [PATCH 1/3] Add overuse detector --- gcc/overuse_detector.go | 69 +++++++++++++++++++++++ gcc/overuse_detector_test.go | 103 +++++++++++++++++++++++++++++++++++ 2 files changed, 172 insertions(+) create mode 100644 gcc/overuse_detector.go create mode 100644 gcc/overuse_detector_test.go diff --git a/gcc/overuse_detector.go b/gcc/overuse_detector.go new file mode 100644 index 0000000..409b797 --- /dev/null +++ b/gcc/overuse_detector.go @@ -0,0 +1,69 @@ +// SPDX-FileCopyrightText: 2025 The Pion community +// SPDX-License-Identifier: MIT + +package gcc + +import ( + "time" +) + +const ( + defaultOveruseTimeThreshold = 5 * time.Millisecond + defaultDelayThreshold = 1.5 +) + +type overuseDetector struct { + overUseTimeThreshold time.Duration + delayThreshold float64 + lastUpdate time.Time + firstOverUse time.Time + overUseCounter int + previousTrend float64 + usage usage +} + +func newOveruseDetector() *overuseDetector { + return &overuseDetector{ + overUseTimeThreshold: defaultOveruseTimeThreshold, + delayThreshold: defaultDelayThreshold, + lastUpdate: time.Time{}, + firstOverUse: time.Time{}, + overUseCounter: 0, + previousTrend: 0, + usage: usageNormal, + } +} + +func (d *overuseDetector) update(ts time.Time, trend float64) usage { + if d.lastUpdate.IsZero() { + d.lastUpdate = ts + } + + switch { + case trend > d.delayThreshold: + if d.firstOverUse.IsZero() { + delta := ts.Sub(d.lastUpdate) + d.firstOverUse = ts.Add(-delta / 2) + } + d.overUseCounter++ + if ts.Sub(d.firstOverUse) > d.overUseTimeThreshold && + d.overUseCounter > 1 && + trend >= d.previousTrend { + d.firstOverUse = time.Time{} + d.overUseCounter = 0 + d.usage = usageOver + } + case trend < -d.delayThreshold: + d.firstOverUse = time.Time{} + d.overUseCounter = 0 + d.usage = usageUnder + default: + d.firstOverUse = time.Time{} + d.overUseCounter = 0 + d.usage = usageNormal + } + d.previousTrend = trend + d.lastUpdate = ts + + return d.usage +} diff --git a/gcc/overuse_detector_test.go b/gcc/overuse_detector_test.go new file mode 100644 index 0000000..dadeb04 --- /dev/null +++ b/gcc/overuse_detector_test.go @@ -0,0 +1,103 @@ +// SPDX-FileCopyrightText: 2025 The Pion community +// SPDX-License-Identifier: MIT + +package gcc + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestOveruseDetectorUpdate(t *testing.T) { + type estimate struct { + ts time.Time + modifiedTrend float64 + } + cases := []struct { + name string + values []estimate + expected []usage + }{ + { + name: "noEstimateNoUsage", + values: []estimate{}, + expected: []usage{}, + }, + { + name: "confirmsOverUse", + values: []estimate{ + {time.Time{}, 0}, + {time.Time{}.Add(5 * time.Millisecond), 40}, + {time.Time{}.Add(20 * time.Millisecond), 90}, + }, + expected: []usage{usageNormal, usageNormal, usageOver}, + }, + { + name: "normaluse", + values: []estimate{{modifiedTrend: 0}}, + expected: []usage{usageNormal}, + }, + { + name: "confirmsUnderUse", + values: []estimate{{time.Time{}, -40}}, + expected: []usage{usageUnder}, + }, + { + name: "noOverUseBeforeDelay", + values: []estimate{ + {time.Time{}.Add(time.Millisecond), 0}, + {time.Time{}.Add(2 * time.Millisecond), 60}, + {time.Time{}.Add(30 * time.Millisecond), 150}, + }, + expected: []usage{usageNormal, usageNormal, usageOver}, + }, + { + name: "noNewOverUseIfEstimateDecreased", + values: []estimate{ + {time.Time{}.Add(time.Millisecond), 0}, + {time.Time{}.Add(10 * time.Millisecond), 80}, + {time.Time{}.Add(30 * time.Millisecond), 60}, + }, + expected: []usage{usageNormal, usageNormal, usageNormal}, + }, + { + name: "normalUseWhenTrendFallsBelowThreshold", + values: []estimate{ + {time.Time{}.Add(time.Millisecond), 0}, + {time.Time{}.Add(10 * time.Millisecond), 80}, + {time.Time{}.Add(30 * time.Millisecond), 150}, + {time.Time{}.Add(35 * time.Millisecond), 1.2}, + }, + expected: []usage{usageNormal, usageNormal, usageOver, usageNormal}, + }, + { + name: "keepsOverUseWhileTrendStaysHigh", + values: []estimate{ + {time.Time{}.Add(5 * time.Millisecond), 0}, + {time.Time{}.Add(10 * time.Millisecond), 1}, + {time.Time{}.Add(15 * time.Millisecond), 1.5}, + {time.Time{}.Add(20 * time.Millisecond), 2}, + {time.Time{}.Add(25 * time.Millisecond), 2.5}, + {time.Time{}.Add(30 * time.Millisecond), 3}, + {time.Time{}.Add(35 * time.Millisecond), 3.5}, + }, + expected: []usage{ + usageNormal, usageNormal, usageNormal, usageNormal, + usageOver, usageOver, usageOver, + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + od := newOveruseDetector() + received := []usage{} + for _, e := range tc.values { + u := od.update(e.ts, e.modifiedTrend) + received = append(received, u) + } + assert.Equal(t, tc.expected, received) + }) + } +} From cae8f4958e56f2a29d4437228dbded771c00fd6e Mon Sep 17 00:00:00 2001 From: Mathis Engelbart Date: Tue, 10 Mar 2026 16:08:39 +0100 Subject: [PATCH 2/3] Add delay rate controller --- gcc/delay_rate_controller.go | 189 +++++++++++++++++++++++++++ gcc/delay_rate_controller_test.go | 205 ++++++++++++++++++++++++++++++ 2 files changed, 394 insertions(+) create mode 100644 gcc/delay_rate_controller.go create mode 100644 gcc/delay_rate_controller_test.go diff --git a/gcc/delay_rate_controller.go b/gcc/delay_rate_controller.go new file mode 100644 index 0000000..4657cc1 --- /dev/null +++ b/gcc/delay_rate_controller.go @@ -0,0 +1,189 @@ +// SPDX-FileCopyrightText: 2025 The Pion community +// SPDX-License-Identifier: MIT + +package gcc + +import ( + "math" + "time" + + "github.com/pion/logging" +) + +const ( + defaultDecreaseFactor = 0.85 + + // minNumDeltas is the number of samples at which the trend is trusted in + // full. Below it the trend is scaled down, so that an estimate based on + // little evidence cannot reach the overuse threshold on its own. + minNumDeltas = 60 +) + +type delayRateController struct { + log logging.LeveledLogger + decreaseFactor float64 + arrivalGroups *arrivalGroupAccumulator + lastArrivalGroup arrivalGroup + trend *trendlineEstimator + overuse *overuseDetector + samples int + usage usage + usageUpdated bool + state state + lastDecreaseRate *ewma + lastUpdate time.Time + targetRate int + minTarget int + maxTarget int +} + +func newDelayRateController(initialRate, minRate, maxRate int, logger logging.LeveledLogger) *delayRateController { + return &delayRateController{ + log: logger, + decreaseFactor: defaultDecreaseFactor, + arrivalGroups: newArrivalGroupAccumulator(), + lastArrivalGroup: []arrivalGroupItem{}, + trend: newTrendlineEstimator(), + overuse: newOveruseDetector(), + usage: 0, + usageUpdated: false, + samples: 0, + state: 0, + lastDecreaseRate: newEWMA(0.95), + targetRate: initialRate, + minTarget: minRate, + maxTarget: maxRate, + } +} + +func (c *delayRateController) onPacketAcked(sequenceNumber uint64, size int, departure, arrival time.Time) { + next := c.arrivalGroups.onPacketAcked( + sequenceNumber, + size, + departure, + arrival, + ) + if next == nil { + return + } + if len(next) == 0 { + // ignore empty groups, should never occur + return + } + if len(c.lastArrivalGroup) == 0 { + c.lastArrivalGroup = next + + return + } + + groupDeparture := next[len(next)-1].Departure + groupArrival := next[len(next)-1].Arrival + lastGroupDeparture := c.lastArrivalGroup[len(c.lastArrivalGroup)-1].Departure + lastGroupArrival := c.lastArrivalGroup[len(c.lastArrivalGroup)-1].Arrival + + interArrivalTime := groupArrival.Sub(lastGroupArrival) + interDepartureTime := groupDeparture.Sub(lastGroupDeparture) + + interGroupDelay := interArrivalTime - interDepartureTime + + trend := c.trend.update(groupArrival, interGroupDelay) + c.samples++ + + modifiedTrend := 0.0 + if c.samples >= 2 { + modifiedTrend = float64(min(c.samples, minNumDeltas)) * trend + } + c.usage = c.overuse.update(groupArrival, modifiedTrend) + c.usageUpdated = true + c.lastArrivalGroup = next + + c.log.Tracef( + "ts=%v.%06d, seq=%v, interArrivalTime=%v, interDepartureTime=%v, interGroupDelay=%v, estimate=%f, modifiedTrend=%f, threshold=%f, usage=%v, state=%v", // nolint + c.lastArrivalGroup[0].Departure.UTC().Format("2006/01/02 15:04:05"), + c.lastArrivalGroup[0].Departure.UTC().Nanosecond()/1e3, + next[0].SequenceNumber, + interArrivalTime.Microseconds(), + interDepartureTime.Microseconds(), + interGroupDelay.Microseconds(), + trend, + modifiedTrend, + c.overuse.delayThreshold, + int(c.usage), + int(c.state), + ) +} + +func (c *delayRateController) update(ts time.Time, deliveryRate int, rtt time.Duration) int { + // The usage is only updated when an arrival group completes. Without a new + // one there is no new delay signal, and running the state machine again + // would act on the previous one twice. + if !c.usageUpdated { + return c.clampedTarget() + } + c.usageUpdated = false + + deliveredRate := float64(deliveryRate) + c.state = c.state.transition(c.usage) + window := ts.Sub(c.lastUpdate) + c.lastUpdate = ts + + // Both the increase cap and the decrease are relative to the rate that was + // actually delivered. Without an estimate (too few acknowledgements in the + // delivery rate window) there is nothing to base a new target on, so keep + // the current target instead of collapsing to the minimum. + if deliveryRate <= 0 { + return c.clampedTarget() + } + + if c.state == stateIncrease { + if c.canIncreaseMultiplicatively(deliveredRate) { + c.targetRate = max(c.targetRate, multiplicativeIncrease(c.targetRate, window)) + } else { + c.targetRate = additiveIncrease(c.targetRate, rtt, window) + } + c.targetRate = min(c.targetRate, int(1.5*deliveredRate)) + } + if c.state == stateDecrease { + c.lastDecreaseRate.update(deliveredRate) + c.targetRate = int(c.decreaseFactor * deliveredRate) + } + + return c.clampedTarget() +} + +// clampedTarget limits the target rate to the configured min and max rates. +func (c *delayRateController) clampedTarget() int { + c.targetRate = max(c.targetRate, c.minTarget) + c.targetRate = min(c.targetRate, c.maxTarget) + + return c.targetRate +} + +func (c *delayRateController) canIncreaseMultiplicatively(deliveredRate float64) bool { + avg := c.lastDecreaseRate.avg() + if avg == 0 { + return true + } + stdDev := math.Sqrt(c.lastDecreaseRate.varr()) + lower := avg - 3*stdDev + upper := avg + 3*stdDev + + return deliveredRate < lower || deliveredRate > upper +} + +func multiplicativeIncrease(rate int, window time.Duration) int { + exponent := min(window.Seconds(), 1.0) + eta := math.Pow(1.08, exponent) + + return int(eta * float64(rate)) +} + +func additiveIncrease(rate int, rtt, window time.Duration) int { + responseTime := 100 + rtt.Milliseconds() + alpha := 0.5 * min(float64(window.Milliseconds())/float64(responseTime), 1.0) + bitsPerFrame := float64(rate) / 30.0 + packetsPerFrame := math.Ceil(bitsPerFrame / (1200 * 8)) + expectedPacketSizeBits := bitsPerFrame / packetsPerFrame + + return rate + max(1000, int(alpha*float64(expectedPacketSizeBits))) +} diff --git a/gcc/delay_rate_controller_test.go b/gcc/delay_rate_controller_test.go new file mode 100644 index 0000000..df2b10e --- /dev/null +++ b/gcc/delay_rate_controller_test.go @@ -0,0 +1,205 @@ +// SPDX-FileCopyrightText: 2025 The Pion community +// SPDX-License-Identifier: MIT + +package gcc + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestDelayRateController(t *testing.T) { + t.Run("init", func(t *testing.T) { + controller := newDelayRateController(1_000_000, 500_000, 2_000_000, nil) + assert.Nil(t, controller.log) + assert.Equal(t, controller.decreaseFactor, defaultDecreaseFactor) + assert.NotNil(t, controller.arrivalGroups) + assert.NotNil(t, controller.lastArrivalGroup) + assert.NotNil(t, controller.trend) + assert.NotNil(t, controller.overuse) + assert.Equal(t, controller.samples, 0) + assert.Equal(t, controller.usage, usage(0)) + assert.Equal(t, controller.state, state(0)) + assert.NotNil(t, controller.lastDecreaseRate) + assert.Zero(t, controller.lastUpdate) + assert.Equal(t, controller.minTarget, 500_000) + assert.Equal(t, controller.maxTarget, 2_000_000) + assert.Equal(t, controller.targetRate, 1_000_000) + }) + + t.Run("canIncreaseMultiplicatively", func(t *testing.T) { + cases := []struct { + deliveredRate float64 + decreaseRate ewma + expected bool + }{ + {deliveredRate: 1000, decreaseRate: ewma{average: 0, variance: 0}, expected: true}, + {deliveredRate: 1000, decreaseRate: ewma{average: 1500, variance: 100}, expected: true}, + {deliveredRate: 1000, decreaseRate: ewma{average: 1020, variance: 100}, expected: false}, + {deliveredRate: 1000, decreaseRate: ewma{average: 800, variance: 50}, expected: true}, + {deliveredRate: 1000, decreaseRate: ewma{average: 995, variance: 100}, expected: false}, + } + + for i, c := range cases { + t.Run(fmt.Sprintf("%v", i), func(t *testing.T) { + controller := newDelayRateController(1000, 500, 2000, nil) + controller.lastDecreaseRate = &c.decreaseRate + assert.Equal(t, c.expected, controller.canIncreaseMultiplicatively(c.deliveredRate)) + }) + } + }) + + t.Run("update", func(t *testing.T) { + cases := []struct { + name string + usage usage + initialRate int + minRate int + maxRate int + deliveryRate int + expected int + }{ + { + name: "hold_on_underuse", + usage: usageUnder, + initialRate: 1_000_000, + minRate: 500_000, + maxRate: 2_000_000, + deliveryRate: 1_000_000, + expected: 1_000_000, + }, + { + name: "no_delivery_rate_holds_target_on_increase", + usage: usageNormal, + initialRate: 1_000_000, + minRate: 500_000, + maxRate: 2_000_000, + deliveryRate: 0, + expected: 1_000_000, + }, + { + name: "no_delivery_rate_holds_target_on_decrease", + usage: usageOver, + initialRate: 1_000_000, + minRate: 500_000, + maxRate: 2_000_000, + deliveryRate: 0, + expected: 1_000_000, + }, + { + name: "multiplicative_increase", + usage: usageNormal, + initialRate: 1_000_000, + minRate: 500_000, + maxRate: 2_000_000, + deliveryRate: 1_000_000, + expected: 1_080_000, + }, + { + name: "increase_capped_at_1.5x_delivered", + usage: usageNormal, + initialRate: 1_000_000, + minRate: 500_000, + maxRate: 2_000_000, + deliveryRate: 600_000, + expected: 900_000, + }, + { + name: "increase_clamped_at_max", + usage: usageNormal, + initialRate: 1_990_000, + minRate: 500_000, + maxRate: 2_000_000, + deliveryRate: 10_000_000, + expected: 2_000_000, + }, + { + name: "decrease_relative_to_delivered", + usage: usageOver, + initialRate: 1_000_000, + minRate: 500_000, + maxRate: 2_000_000, + deliveryRate: 1_000_000, + expected: 850_000, + }, + { + name: "decrease_clamped_at_min", + usage: usageOver, + initialRate: 1_000_000, + minRate: 500_000, + maxRate: 2_000_000, + deliveryRate: 200_000, + expected: 500_000, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + controller := newDelayRateController(tc.initialRate, tc.minRate, tc.maxRate, nil) + controller.usage = tc.usage + controller.usageUpdated = true + now := time.Time{}.Add(time.Hour) + controller.lastUpdate = now.Add(-time.Second) + + res := controller.update(now, tc.deliveryRate, 100*time.Millisecond) + assert.InDelta(t, tc.expected, res, 1) + }) + } + }) + + t.Run("updateWithoutNewArrivalGroup", func(t *testing.T) { + controller := newDelayRateController(1_000_000, 500_000, 2_000_000, nil) + controller.usage = usageOver + controller.usageUpdated = true + now := time.Time{}.Add(time.Hour) + controller.lastUpdate = now.Add(-time.Second) + + // The first update acts on the overuse. + assert.Equal(t, 850_000, controller.update(now, 1_000_000, 100*time.Millisecond)) + + // The sender follows the lowered target, so the delivery rate drops with + // it. No arrival group completed since, so there is no new delay signal + // and the same overuse must not decrease the target again. + now = now.Add(100 * time.Millisecond) + assert.Equal(t, 850_000, controller.update(now, 850_000, 100*time.Millisecond)) + now = now.Add(100 * time.Millisecond) + assert.Equal(t, 850_000, controller.update(now, 722_500, 100*time.Millisecond)) + }) + + t.Run("multiplicativeIncrease", func(t *testing.T) { + cases := []struct { + initialRate int + rate int + window time.Duration + expected float64 + }{ + {initialRate: 1000, rate: 1000, window: 100 * time.Millisecond, expected: 1007}, + } + for i, c := range cases { + t.Run(fmt.Sprintf("%v", i), func(t *testing.T) { + res := multiplicativeIncrease(c.rate, c.window) + assert.InDelta(t, res, c.expected, 1) + }) + } + }) + + t.Run("additiveIncrease", func(t *testing.T) { + cases := []struct { + initialRate int + rate int + window time.Duration + expected int + }{ + {initialRate: 1000, rate: 1000, window: 100 * time.Millisecond, expected: 2000}, + {initialRate: 1_000_000, rate: 1_500_000, window: 100 * time.Millisecond, expected: 1_500_000 + 2083}, + } + for i, c := range cases { + t.Run(fmt.Sprintf("%v", i), func(t *testing.T) { + res := additiveIncrease(c.rate, 100*time.Millisecond, c.window) + assert.InDelta(t, res, c.expected, 1) + }) + } + }) +} From c796323cd815cb98a4abefe354dfb3722e7d9e7e Mon Sep 17 00:00:00 2001 From: Mathis Engelbart Date: Mon, 27 Apr 2026 17:41:40 +0200 Subject: [PATCH 3/3] Add send side BWE controller --- gcc/send_side_bwe.go | 96 +++++++++++++++++++++++++++++++ gcc/send_side_bwe_test.go | 116 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 212 insertions(+) create mode 100644 gcc/send_side_bwe.go create mode 100644 gcc/send_side_bwe_test.go diff --git a/gcc/send_side_bwe.go b/gcc/send_side_bwe.go new file mode 100644 index 0000000..5178fb5 --- /dev/null +++ b/gcc/send_side_bwe.go @@ -0,0 +1,96 @@ +// SPDX-FileCopyrightText: 2025 The Pion community +// SPDX-License-Identifier: MIT + +package gcc + +import ( + "time" + + "github.com/pion/logging" +) + +// Option is a functional option for a SendSideController. +type Option func(*SendSideController) error + +// WithLoggerFactory configures a custom logger factory for a +// SendSideController. +func WithLoggerFactory(lf logging.LoggerFactory) Option { + return func(ssc *SendSideController) error { + ssc.logFactory = lf + + return nil + } +} + +// SendSideController is a sender side congestion controller. +// +// A SendSideController must not be used concurrently from multiple goroutines. +type SendSideController struct { + logFactory logging.LoggerFactory + log logging.LeveledLogger + dre *deliveryRateEstimator + lrc *lossRateController + drc *delayRateController + targetRate int +} + +// NewSendSideController creates a new SendSideController with initial, min and +// max rates. +func NewSendSideController(initialRate, minRate, maxRate int, opts ...Option) (*SendSideController, error) { + ssc := &SendSideController{ + logFactory: logging.NewDefaultLoggerFactory(), + dre: newDeliveryRateEstimator(time.Second), + lrc: newLossRateController(initialRate, minRate, maxRate), + targetRate: initialRate, + } + for _, opt := range opts { + if err := opt(ssc); err != nil { + return nil, err + } + } + ssc.log = ssc.logFactory.NewLogger("bwe_send_side_controller") + ssc.drc = newDelayRateController(initialRate, minRate, maxRate, ssc.logFactory.NewLogger("bwe_delay_rate_controller")) + + return ssc, nil +} + +// OnLoss must be called when a packet is reported as lost. Packets MUST not be +// reported more than once. +func (c *SendSideController) OnLoss() { + c.lrc.onPacketLost() +} + +// OnAck must be called when new acknowledgments arrive. Packets MUST not be +// acknowledged more than once. +func (c *SendSideController) OnAck(sequenceNumber uint64, size int, departure, arrival time.Time) { + c.lrc.onPacketAcked() + if !arrival.IsZero() { + c.dre.onPacketAcked(arrival, size) + c.drc.onPacketAcked( + sequenceNumber, + size, + departure, + arrival, + ) + } +} + +// OnFeedback must be called when a new feedback report arrives. ts is the +// arrival timestamp of the feedback report. rtt is the latest RTT sample. It +// returns the new target rate. +func (c *SendSideController) OnFeedback(ts time.Time, rtt time.Duration) int { + delivered := c.dre.getRate() + lossTarget := c.lrc.update(delivered) + delayTarget := c.drc.update(ts, delivered, rtt) + c.targetRate = min(lossTarget, delayTarget) + c.log.Tracef( + "rtt=%v, delivered=%v, lossTarget=%v, delayTarget=%v, target=%v", + rtt.Nanoseconds(), + delivered, + lossTarget, + delayTarget, + c.targetRate, + ) + + return c.targetRate +} diff --git a/gcc/send_side_bwe_test.go b/gcc/send_side_bwe_test.go new file mode 100644 index 0000000..8cb3462 --- /dev/null +++ b/gcc/send_side_bwe_test.go @@ -0,0 +1,116 @@ +// SPDX-FileCopyrightText: 2026 The Pion community +// SPDX-License-Identifier: MIT + +package gcc + +import ( + "errors" + "testing" + "time" + + "github.com/pion/logging" + "github.com/stretchr/testify/assert" +) + +var errTestOption = errors.New("test option") + +func TestSendSideController(t *testing.T) { + t.Run("init", func(t *testing.T) { + ssc, err := NewSendSideController(1_000_000, 100_000, 2_000_000) + assert.NoError(t, err) + assert.NotNil(t, ssc.logFactory) + assert.NotNil(t, ssc.log) + assert.NotNil(t, ssc.dre) + assert.NotNil(t, ssc.lrc) + assert.NotNil(t, ssc.drc) + assert.Equal(t, 1_000_000, ssc.targetRate) + }) + + t.Run("withLoggerFactory", func(t *testing.T) { + factory := logging.NewDefaultLoggerFactory() + ssc, err := NewSendSideController(1_000_000, 100_000, 2_000_000, WithLoggerFactory(factory)) + assert.NoError(t, err) + assert.Same(t, factory, ssc.logFactory) + }) + + t.Run("optionError", func(t *testing.T) { + ssc, err := NewSendSideController(1_000_000, 100_000, 2_000_000, func(*SendSideController) error { + return errTestOption + }) + assert.ErrorIs(t, err, errTestOption) + assert.Nil(t, ssc) + }) + + t.Run("noFeedbackKeepsTargetRate", func(t *testing.T) { + ssc, err := NewSendSideController(1_000_000, 100_000, 2_000_000) + assert.NoError(t, err) + assert.Equal(t, 1_000_000, ssc.OnFeedback(time.Time{}.Add(time.Hour), 100*time.Millisecond)) + }) + + t.Run("targetIsMinimumOfLossAndDelayTarget", func(t *testing.T) { + ssc, err := NewSendSideController(1_000_000, 100_000, 2_000_000) + assert.NoError(t, err) + + // A single lost packet is a loss rate of 100%, so the loss controller + // halves its target. The delay controller has not seen any arrival + // group and keeps the initial rate, so the loss target wins. + ssc.OnLoss() + rate := ssc.OnFeedback(time.Time{}.Add(time.Hour), 100*time.Millisecond) + assert.Equal(t, 500_000, rate) + assert.Equal(t, rate, ssc.targetRate) + }) + + t.Run("ackWithoutArrivalOnlyCountsForLoss", func(t *testing.T) { + ssc, err := NewSendSideController(1_000_000, 100_000, 2_000_000) + assert.NoError(t, err) + + ssc.OnAck(0, 1200, time.Time{}.Add(time.Hour), time.Time{}) + assert.Equal(t, 1, ssc.lrc.packetsSinceLastUpdate) + assert.Zero(t, ssc.dre.getRate()) + assert.Zero(t, ssc.drc.samples) + }) + + t.Run("increasesWhileDelivering", func(t *testing.T) { + ssc, err := NewSendSideController(1_000_000, 100_000, 2_000_000) + assert.NoError(t, err) + + // 100 packets at a steady rate and a constant one way delay: no queue + // builds up, so both controllers are free to increase. + departure := time.Time{}.Add(time.Hour) + arrival := departure.Add(50 * time.Millisecond) + for i := range 100 { + offset := time.Duration(i) * time.Millisecond + ssc.OnAck(uint64(i), 1200, departure.Add(offset), arrival.Add(offset)) // nolint:gosec // loop index + } + + rate := ssc.OnFeedback(departure.Add(150*time.Millisecond), 100*time.Millisecond) + assert.Greater(t, rate, 1_000_000) + assert.Equal(t, rate, ssc.targetRate) + }) + + t.Run("decreasesWhileDelayIncreases", func(t *testing.T) { + ssc, err := NewSendSideController(8_000_000, 100_000, 10_000_000) + assert.NoError(t, err) + + // Packets depart every millisecond but arrive 1.5 milliseconds apart, so + // a queue is building up and the overuse detector reports an overuse. + departure := time.Time{}.Add(time.Hour) + arrival := departure.Add(50 * time.Millisecond) + for i := range 100 { + ssc.OnAck( + uint64(i), // nolint:gosec // loop index + 1200, + departure.Add(time.Duration(i)*time.Millisecond), + arrival.Add(time.Duration(i)*1500*time.Microsecond), + ) + } + + // 99 packets delivered over 148.5ms is 6.4Mbps, and the delay controller + // decreases to 85% of that. + rate := ssc.OnFeedback(departure.Add(200*time.Millisecond), 100*time.Millisecond) + assert.Equal(t, usageOver, ssc.drc.usage) + assert.Equal(t, stateDecrease, ssc.drc.state) + assert.Equal(t, 5_440_000, rate) + assert.Equal(t, rate, ssc.targetRate) + }) +}