fix: stop adaptive pacing from compounding its own rate limit

docs/ACCELERATION.md

@@ -78,6 +78,26 @@ backpressure cannot be hidden by small idle gaps. Subsequent active samples can
 still reduce the target when capacity, RTT or loss changes. Fixed-rate and bypass
 modes are unchanged.
 
+The controller separately measures time actually spent waiting for missing
+tokens. Only one admission owner can sleep at a time, so concurrent writers do
+not multiply that duration; observations include an in-progress sleep and use
+the controller's own clock for its cumulative wait baseline. When this waiting
+occupies at least half a valid observation interval, a lower observed rate does
+not replace the unpenalized capacity history. Higher observations are still
+accepted. For valid delivery observations, current RTT and loss factors still
+apply to that capacity, not to a rate already reduced by the previous penalty. With no
+capacity history yet, a pacing-limited sample uses the configured initial rate
+as a prior; an unconstrained sample can replace it with a lower capacity.
+
+This qualification never subtracts waiting time from the delivery-rate
+denominator: QUIC can send already-queued bytes during an application wait.
+Actual transport backpressure remains outside token sleep. When a slower path
+fills the finite send buffers, predominantly transport-bound samples can age
+out the old bandwidth maximum. Sleep share is still an application-level
+heuristic, not proof that the path is uncongested. The controller does not add
+a separate capacity-probing phase: sustained RTT penalties can limit discovery
+of spare capacity until conditions or higher delivery observations change.
+
 This is intentionally not a full BBR state machine. In particular AutoCAR has
 no transport-visible BDP congestion window, ACK aggregation model, ProbeRTT
 drain, ECN policy, inflight bounds or BBRv2/BBRv3 logic. Calling it “real BBR”

docs/ARCHITECTURE.md

@@ -212,6 +212,13 @@ bandwidth estimate. Predominantly busy intervals remain eligible. Pending
 pacing waits and blocked transport writes are not idle; concurrent streams
 share the activity accounting and ordered counter observations.
 
+The controller also records its single admission owner's actual token sleeps.
+In pacing-dominated intervals, lower observations do not overwrite capacity
+history, but current RTT/loss penalties still apply. Transport-bound intervals
+can lower the estimate. This prevents an application-imposed rate reduction
+from repeatedly becoming the next capacity estimate; it neither bypasses
+QUIC congestion control nor infers wire capacity by subtracting sleep time.
+
 Web H2/H3 streams do not use the native pacing negotiation and report client
 and relay pacing as `not-applicable`. Fixed-rate is rejected with
 `web-auto`, `h3`, and `h2`.

internal/accel/adaptive.go

@@ -22,13 +22,14 @@ type adaptiveSettings struct {
 // deliberately smaller than a transport congestion controller and has no
 // control over the transport's congestion window or retransmission behavior.
 type adaptiveEstimator struct {
-	settings adaptiveSettings
-	rates    [deliveryRateWindow]float64
-	next     int
-	count    int
+	settings    adaptiveSettings
+	initialRate float64
+	rates       [deliveryRateWindow]float64
+	next        int
+	count       int
 }
 
-func newAdaptiveEstimator(profile Profile) adaptiveEstimator {
+func newAdaptiveEstimator(profile Profile, initialRate int64) adaptiveEstimator {
 	settings := adaptiveSettings{}
 	switch profile {
 	case ProfileConservative:
@@ -59,7 +60,7 @@ func newAdaptiveEstimator(profile Profile) adaptiveEstimator {
 			minimumLossFactor: 0.60,
 		}
 	}
-	return adaptiveEstimator{settings: settings}
+	return adaptiveEstimator{settings: settings, initialRate: float64(initialRate)}
 }
 
 func (e *adaptiveEstimator) reset() {
@@ -73,17 +74,26 @@ func (e *adaptiveEstimator) observe(
 	lossRatio float64,
 	minimumRTT time.Duration,
 	smoothedRTT time.Duration,
+	pacingLimited bool,
 ) float64 {
-	e.rates[e.next] = math.Max(0, deliveredRate)
-	e.next = (e.next + 1) % len(e.rates)
-	if e.count < len(e.rates) {
-		e.count++
+	deliveredRate = math.Max(0, deliveredRate)
+	bottleneckRate := e.bandwidthEstimate()
+	if !pacingLimited || deliveredRate > bottleneckRate {
+		// An application pacing limit censors lower capacity observations. Do
+		// not let those samples replace the unpenalized bandwidth history with
+		// the result of our own previous RTT/loss reduction. Higher observations
+		// remain useful evidence, and unconstrained samples can age out an old
+		// maximum when the path really becomes slower.
+		e.rates[e.next] = deliveredRate
+		e.next = (e.next + 1) % len(e.rates)
+		if e.count < len(e.rates) {
+			e.count++
+		}
+		bottleneckRate = e.bandwidthEstimate()
 	}
 
-	bottleneckRate := float64(0)
-	for index := 0; index < e.count; index++ {
-		bottleneckRate = math.Max(bottleneckRate, e.rates[index])
-	}
+	// Always apply current congestion signals to capacity evidence, even when
+	// a pacing-limited sample was not allowed to change that evidence.
 	target := bottleneckRate * e.settings.pacingGain
 
 	rttRatio := float64(smoothedRTT) / float64(minimumRTT)
@@ -97,3 +107,16 @@ func (e *adaptiveEstimator) observe(
 	}
 	return target
 }
+
+func (e *adaptiveEstimator) bandwidthEstimate() float64 {
+	if e.count == 0 {
+		// This is only a prior, not a permanent minimum or a history entry. The
+		// first non-limited observation may establish a lower path capacity.
+		return e.initialRate
+	}
+	maximum := float64(0)
+	for index := 0; index < e.count; index++ {
+		maximum = math.Max(maximum, e.rates[index])
+	}
+	return maximum
+}

internal/accel/adaptive_feedback_test.go

@@ -0,0 +1,136 @@
+package accel
+
+import (
+	"math"
+	"testing"
+	"time"
+)
+
+func TestAdaptiveEstimatorLimitedSamplesPreserveCapacityHistory(t *testing.T) {
+	for _, profile := range []Profile{ProfileConservative, ProfileBalanced, ProfileAggressive} {
+		t.Run(profile.String(), func(t *testing.T) {
+			estimator := newAdaptiveEstimator(profile, 2_000_000)
+			estimator.observe(8_000_000, 0, time.Millisecond, time.Millisecond, false)
+			learned := estimator
+			for range 4 * deliveryRateWindow {
+				estimator.observe(100_000, 0, time.Millisecond, time.Millisecond, true)
+				if estimator != learned {
+					t.Fatal("low pacing-limited sample aged or changed capacity history")
+				}
+			}
+			estimator.observe(8_000_000, 0, time.Millisecond, time.Millisecond, true)
+			if estimator != learned {
+				t.Fatal("equal pacing-limited sample aged capacity history")
+			}
+			got := estimator.observe(9_000_000, 0, time.Millisecond, time.Millisecond, true)
+			assertEstimatorRate(t, got, 9_000_000*estimator.settings.pacingGain)
+			if estimator.count != learned.count+1 || estimator.bandwidthEstimate() != 9_000_000 {
+				t.Fatal("higher pacing-limited sample was not learned")
+			}
+		})
+	}
+}
+
+func TestAdaptiveEstimatorUnconstrainedCapacityCanFallAndRecover(t *testing.T) {
+	for _, profile := range []Profile{ProfileConservative, ProfileBalanced, ProfileAggressive} {
+		t.Run(profile.String(), func(t *testing.T) {
+			estimator := newAdaptiveEstimator(profile, 8_000_000)
+			estimator.observe(8_000_000, 0, time.Millisecond, time.Millisecond, false)
+			for index := 0; index < deliveryRateWindow; index++ {
+				got := estimator.observe(1_000_000, 0, time.Millisecond, time.Millisecond, false)
+				wantCapacity := float64(8_000_000)
+				if index == deliveryRateWindow-1 {
+					wantCapacity = 1_000_000
+				}
+				assertEstimatorRate(t, got, wantCapacity*estimator.settings.pacingGain)
+			}
+			got := estimator.observe(4_000_000, 0, time.Millisecond, time.Millisecond, false)
+			assertEstimatorRate(t, got, 4_000_000*estimator.settings.pacingGain)
+		})
+	}
+}
+
+func TestAdaptiveEstimatorLimitedCongestionDoesNotCompound(t *testing.T) {
+	for _, test := range []struct {
+		profile   Profile
+		gain      float64
+		rttFactor float64
+		loss      float64
+	}{
+		{ProfileConservative, 1.00, 0.575, 0.62},
+		{ProfileBalanced, 1.08, 0.65, 0.745},
+		{ProfileAggressive, 1.18, 0.75, 0.86},
+	} {
+		t.Run(test.profile.String(), func(t *testing.T) {
+			const capacity = 8_000_000
+			estimator := newAdaptiveEstimator(test.profile, capacity)
+			current := estimator.observe(capacity, 0, time.Millisecond, 2*time.Millisecond, false)
+			assertEstimatorRate(t, current, capacity*test.gain*test.rttFactor)
+			learned := estimator
+			for range 4 * deliveryRateWindow {
+				current = estimator.observe(current, 0, time.Millisecond, 2*time.Millisecond, true)
+				assertEstimatorRate(t, current, capacity*test.gain*test.rttFactor)
+			}
+			for range 4 * deliveryRateWindow {
+				current = estimator.observe(current, 0.2, time.Millisecond, 2*time.Millisecond, true)
+				assertEstimatorRate(t, current, capacity*test.gain*test.rttFactor*test.loss)
+			}
+			worseLoss := estimator.observe(current, 0.4, time.Millisecond, 2*time.Millisecond, true)
+			if worseLoss >= current {
+				t.Fatalf("worsening loss did not lower target: %g >= %g", worseLoss, current)
+			}
+			worseRTT := estimator.observe(current, 0.2, time.Millisecond, 4*time.Millisecond, true)
+			if worseRTT >= current {
+				t.Fatalf("worsening RTT did not lower target: %g >= %g", worseRTT, current)
+			}
+			recovered := estimator.observe(worseRTT, 0, time.Millisecond, time.Millisecond, true)
+			assertEstimatorRate(t, recovered, capacity*test.gain)
+			if estimator != learned {
+				t.Fatal("RTT/loss changes changed retained bandwidth evidence")
+			}
+		})
+	}
+}
+
+func TestAdaptiveEstimatorEmptyHistoryUsesPriorOnlyWhenLimited(t *testing.T) {
+	for _, profile := range []Profile{ProfileConservative, ProfileBalanced, ProfileAggressive} {
+		t.Run(profile.String(), func(t *testing.T) {
+			const initial = 8_000_000
+			estimator := newAdaptiveEstimator(profile, initial)
+			prior := estimator
+			// The comparison estimator has actual capacity evidence, so its
+			// penalties should match the empty-history prior under the same RTT
+			// and loss without inserting sparse ACK traffic into history.
+			reference := newAdaptiveEstimator(profile, initial)
+			want := reference.observe(initial, 0.2, time.Millisecond, 2*time.Millisecond, false)
+			for range 4 * deliveryRateWindow {
+				got := estimator.observe(100, 0.2, time.Millisecond, 2*time.Millisecond, true)
+				assertEstimatorRate(t, got, want)
+				if estimator != prior {
+					t.Fatal("sparse limited sample initialized bandwidth history")
+				}
+			}
+			got := estimator.observe(10_000_000, 0, time.Millisecond, time.Millisecond, true)
+			assertEstimatorRate(t, got, 10_000_000*estimator.settings.pacingGain)
+			if estimator.count != 1 || estimator.bandwidthEstimate() != 10_000_000 {
+				t.Fatal("higher-than-prior limited observation was not learned")
+			}
+			estimator.reset()
+			if estimator != prior {
+				t.Fatal("reset did not restore empty history while retaining the initial prior")
+			}
+			got = estimator.observe(100_000, 0, time.Millisecond, time.Millisecond, false)
+			assertEstimatorRate(t, got, 100_000*estimator.settings.pacingGain)
+			if estimator.count != 1 || estimator.bandwidthEstimate() != 100_000 {
+				t.Fatal("initial prior prevented learning a truly low capacity")
+			}
+		})
+	}
+}
+
+func assertEstimatorRate(t *testing.T, got, want float64) {
+	t.Helper()
+	if math.Abs(got-want) > 0.000001 {
+		t.Fatalf("estimator target = %g, want %g", got, want)
+	}
+}

internal/accel/pacer.go

@@ -129,6 +129,15 @@ type Controller struct {
 	haveSnapshot bool
 	previous     Snapshot
 	estimator    adaptiveEstimator
+
+	// Only the admission owner sleeps for missing tokens. Count this interval
+	// once, not the overlapping Wait durations of queued writers. Observation
+	// baselines use our own clock, independently of a caller's Snapshot.At.
+	pacingSleepStarted      time.Time
+	pacingSleeping          bool
+	pacingSleepTime         time.Duration
+	previousPacingSleep     time.Duration
+	previousObservationTime time.Time
 }
 
 type normalizedConfig struct {
@@ -187,7 +196,7 @@ func New(config Config) (*Controller, error) {
 		lastRefill:  normalized.clock.Now(),
 		admission:   makeAdmissionToken(),
 		stateChange: make(chan struct{}),
-		estimator:   newAdaptiveEstimator(normalized.profile),
+		estimator:   newAdaptiveEstimator(normalized.profile, normalized.initialRate),
 	}, nil
 }
 
@@ -258,7 +267,7 @@ func (c *Controller) Observe(snapshot Snapshot) error {
 	defer c.mu.Unlock()
 
 	if !c.haveSnapshot {
-		c.previous = snapshot
+		c.recordObservationLocked(snapshot, c.clock.Now())
 		c.haveSnapshot = true
 		return nil
 	}
@@ -269,7 +278,7 @@ func (c *Controller) Observe(snapshot Snapshot) error {
 	if snapshot.SentBytes < c.previous.SentBytes || snapshot.ApplicationIdleTime < c.previous.ApplicationIdleTime {
 		// A cumulative counter reset denotes a new transport epoch. Rebaseline
 		// instead of interpreting wrapped counters as a huge delivery sample.
-		c.previous = snapshot
+		c.recordObservationLocked(snapshot, c.clock.Now())
 		c.estimator.reset()
 		c.setTargetRateLocked(c.initialRate)
 		return nil
@@ -284,13 +293,17 @@ func (c *Controller) Observe(snapshot Snapshot) error {
 		return nil
 	}
 
+	now := c.clock.Now()
+	observationElapsed := now.Sub(c.previousObservationTime)
+	pacingElapsed := c.pacingTimeLocked(now) - c.previousPacingSleep
+	pacingLimited := observationElapsed > 0 && pacingElapsed > 0 && pacingElapsed >= observationElapsed/2
 	idleDelta := snapshot.ApplicationIdleTime - c.previous.ApplicationIdleTime
 	sentDelta := snapshot.SentBytes - c.previous.SentBytes
 	lostDelta := uint64(0)
 	if snapshot.LostBytes >= c.previous.LostBytes {
 		lostDelta = snapshot.LostBytes - c.previous.LostBytes
 	}
-	c.previous = snapshot
+	c.recordObservationLocked(snapshot, now)
 
 	if idleDelta >= window && idleDelta >= elapsed/2 {
 		// ACK/control traffic during application silence does not measure path
@@ -313,12 +326,29 @@ func (c *Controller) Observe(snapshot Snapshot) error {
 	deliveredRate := float64(deliveredDelta) / elapsed.Seconds()
 	lossRatio := float64(lostDelta) / float64(sentDelta)
 
-	target := c.estimator.observe(deliveredRate, lossRatio, snapshot.MinRTT, snapshot.SmoothedRTT)
+	// QUIC can transmit queued bytes while our writer waits for tokens. Do not
+	// subtract pacing time from elapsed to invent a larger delivery rate. It
+	// only qualifies lower capacity samples; RTT/loss still affect the target.
+	target := c.estimator.observe(deliveredRate, lossRatio, snapshot.MinRTT, snapshot.SmoothedRTT, pacingLimited)
 	target = math.Max(float64(c.minimumRate), math.Min(float64(c.maximumRate), target))
 	c.setTargetRateLocked(int64(math.Round(target)))
 	return nil
 }
 
+func (c *Controller) recordObservationLocked(snapshot Snapshot, now time.Time) {
+	c.previous = snapshot
+	c.previousObservationTime = now
+	c.previousPacingSleep = c.pacingTimeLocked(now)
+}
+
+func (c *Controller) pacingTimeLocked(now time.Time) time.Duration {
+	elapsed := c.pacingSleepTime
+	if c.pacingSleeping && now.After(c.pacingSleepStarted) {
+		elapsed += now.Sub(c.pacingSleepStarted)
+	}
+	return elapsed
+}
+
 func stableSampleWindow(minimumRTT time.Duration) time.Duration {
 	window := minimumRTT / 4
 	if window < minimumSampleWindow {
@@ -388,9 +418,21 @@ func (c *Controller) admit(ctx context.Context, amount float64) error {
 		missing := amount - c.tokens
 		delay := durationForBytes(missing, c.targetRate)
 		stateChange := c.stateChange
+		if c.mode == ModeAdaptive {
+			c.pacingSleepStarted = c.clock.Now()
+			c.pacingSleeping = true
+		}
 		c.mu.Unlock()
 
-		if err := c.sleepUntilStateChange(ctx, delay, stateChange); err != nil {
+		err := c.sleepUntilStateChange(ctx, delay, stateChange)
+		if c.mode == ModeAdaptive {
+			c.mu.Lock()
+			c.pacingSleepTime = c.pacingTimeLocked(c.clock.Now())
+			c.pacingSleepStarted = time.Time{}
+			c.pacingSleeping = false
+			c.mu.Unlock()
+		}
+		if err != nil {
 			return err
 		}
 	}