fix: keep idle traffic from collapsing adaptive pacing

docs/ACCELERATION.md

@@ -58,6 +58,26 @@ The target starts from 64 Mbit/s, applies the selected pacing gain, and is then
 reduced when either RTT inflation or interval loss crosses a profile threshold.
 Every target is clamped to configured minimum and maximum rates.
 
+The native adapter also tracks cumulative application-send idle time across
+the whole connection. Stream writes and datagram batches remain active while
+waiting for pacing tokens or QUIC capacity; overlapping sends count as one busy
+interval. If idle time spans both at least one stable sampling window and at
+least half the observed interval, the controller
+rebaselines the counters without changing its target or bandwidth history.
+The window is one quarter of minimum RTT, bounded to 10–250 ms. Cumulative
+accounting keeps idle gaps visible even when several writers sample inside
+that window. This prevents ACK/control traffic during a receive-only interval
+from being mistaken for low outbound capacity when traffic changes direction.
+
+This is a conservative application-idle filter, not transport-level knowledge
+of every queued packet. A discarded interval also does not update the
+application-layer RTT/loss response; the underlying QUIC congestion controller
+remains active throughout. Predominantly busy intervals still update the
+estimate even when a short source gap exceeds one sampling window, so repeated
+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.
+
 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

@@ -205,6 +205,13 @@ apply only to QUIC. A TCP flow taken by `auto`'s TLS fallback is unpaced; explic
 congestion control, packetization, ACK handling and RFC 9002 recovery: upstream
 quic-go for native, and the pinned fork for web H3.
 
+Native adaptive samples include connection-wide application-send idle time.
+Intervals that are at least half idle, with a stable sampling window of no
+pending sends, are rebaselined instead of turning ACK-only traffic into a low
+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.
+
 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/pacer.go

@@ -71,6 +71,11 @@ type Snapshot struct {
 	LostBytes   uint64
 	MinRTT      time.Duration
 	SmoothedRTT time.Duration
+	// ApplicationIdleTime is the cumulative time for which the sender had no
+	// application writes in progress. Waiting for pacing or transport capacity
+	// is active, not idle. A zero value preserves adapters without idle tracking;
+	// a decrease, like a SentBytes reset, starts a new observation epoch.
+	ApplicationIdleTime time.Duration
 }
 
 // Clock provides the time operations used by pacing. Implementations must be
@@ -242,6 +247,9 @@ func (c *Controller) Observe(snapshot Snapshot) error {
 	if snapshot.MinRTT < 0 || snapshot.SmoothedRTT < 0 {
 		return fmt.Errorf("%w: RTT values cannot be negative", ErrInvalidSnapshot)
 	}
+	if snapshot.ApplicationIdleTime < 0 {
+		return fmt.Errorf("%w: application idle time cannot be negative", ErrInvalidSnapshot)
+	}
 	if snapshot.At.IsZero() {
 		snapshot.At = c.clock.Now()
 	}
@@ -258,8 +266,8 @@ func (c *Controller) Observe(snapshot Snapshot) error {
 		return fmt.Errorf("%w: sample time must increase", ErrInvalidSnapshot)
 	}
 
-	if snapshot.SentBytes < c.previous.SentBytes {
-		// A cumulative sent-byte reset denotes a new transport epoch. Rebaseline
+	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.estimator.reset()
@@ -268,20 +276,31 @@ func (c *Controller) Observe(snapshot Snapshot) error {
 	}
 
 	elapsed := snapshot.At.Sub(c.previous.At)
+	window := stableSampleWindow(snapshot.MinRTT)
 	// Connection counters are sampled by every writer. Ignore sub-window calls
 	// without advancing the baseline so concurrent streams cannot turn one packet
 	// observed a few microseconds later into a terabyte-per-second rate sample.
-	if elapsed < stableSampleWindow(snapshot.MinRTT) {
+	if elapsed < window {
 		return nil
 	}
 
+	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
 
+	if idleDelta >= window && idleDelta >= elapsed/2 {
+		// ACK/control traffic during application silence does not measure path
+		// capacity. Keep the learned target and estimator, but rebaseline so
+		// the next active interval does not inherit this idle time. Sub-window
+		// observations above retain both counters until this decision is made.
+		// Require idle time to dominate the interval: a short source gap after
+		// a long, flow-controlled write must not hide genuine path congestion.
+		return nil
+	}
 	if sentDelta == 0 || snapshot.MinRTT == 0 || snapshot.SmoothedRTT == 0 {
 		return nil
 	}

internal/accel/pacing_idle_test.go

@@ -0,0 +1,275 @@
+package accel
+
+import (
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestAdaptiveApplicationIdleACKWindowPreservesTarget(t *testing.T) {
+	controller, err := New(Config{Clock: newFakeClock()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	base := Snapshot{
+		At: time.Unix(1, 0), SentBytes: 1000,
+		MinRTT: time.Millisecond, SmoothedRTT: time.Millisecond,
+	}
+	mustObserveIdleSnapshot(t, controller, base)
+	beforeEstimator := controller.estimator
+	idle := base
+	idle.At = idle.At.Add(2500 * time.Millisecond)
+	idle.SentBytes += 20_000 // ACK/control traffic while receiving an upload.
+	idle.ApplicationIdleTime = 2500 * time.Millisecond
+	mustObserveIdleSnapshot(t, controller, idle)
+	if got := controller.TargetBytesPerSecond(); got != defaultInitialRate {
+		t.Fatalf("ACK-only application-idle window changed target to %d, want %d", got, defaultInitialRate)
+	}
+	if controller.estimator != beforeEstimator {
+		t.Fatal("application-idle window changed bandwidth history")
+	}
+	if controller.previous != idle {
+		t.Fatal("application-idle window did not rebaseline cumulative counters")
+	}
+}
+
+func TestAdaptiveApplicationIdleAccumulatesAcrossSubWindowSamples(t *testing.T) {
+	controller := newTestAdaptive(t, ProfileBalanced)
+	base := Snapshot{At: time.Unix(1, 0), MinRTT: 40 * time.Millisecond, SmoothedRTT: 40 * time.Millisecond}
+	mustObserveIdleSnapshot(t, controller, base)
+	for _, elapsed := range []time.Duration{4 * time.Millisecond, 8 * time.Millisecond} {
+		snapshot := base
+		snapshot.At = base.At.Add(elapsed)
+		snapshot.SentBytes = uint64(elapsed / time.Microsecond)
+		snapshot.ApplicationIdleTime = elapsed
+		mustObserveIdleSnapshot(t, controller, snapshot)
+		if controller.previous != base {
+			t.Fatal("sub-window observation consumed accumulated idle time")
+		}
+	}
+	threshold := base
+	threshold.At = base.At.Add(10 * time.Millisecond)
+	threshold.SentBytes = 10_000
+	threshold.ApplicationIdleTime = 10 * time.Millisecond
+	mustObserveIdleSnapshot(t, controller, threshold)
+	if got := controller.TargetBytesPerSecond(); got != 1_000_000 || controller.estimator.count != 0 {
+		t.Fatalf("accumulated idle window changed target/history: %d/%d", got, controller.estimator.count)
+	}
+	if controller.previous != threshold {
+		t.Fatal("full idle window was not rebaselined")
+	}
+	active := threshold
+	active.At = active.At.Add(20 * time.Millisecond)
+	active.SentBytes += 20_000
+	mustObserveIdleSnapshot(t, controller, active)
+	if got := controller.TargetBytesPerSecond(); got != 1_080_000 {
+		t.Fatalf("clean active interval reused old idle time: target=%d want 1080000", got)
+	}
+}
+
+func TestAdaptiveApplicationShortIdleGapsStillUpdateRate(t *testing.T) {
+	controller, err := New(Config{Clock: newFakeClock()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	base := Snapshot{At: time.Unix(1, 0), MinRTT: time.Millisecond, SmoothedRTT: time.Millisecond}
+	mustObserveIdleSnapshot(t, controller, base)
+	sample := base
+	sample.At = base.At.Add(50 * time.Millisecond)
+	sample.SentBytes = 50_000
+	sample.ApplicationIdleTime = 9 * time.Millisecond
+	mustObserveIdleSnapshot(t, controller, sample)
+	if got := controller.TargetBytesPerSecond(); got != 1_080_000 {
+		t.Fatalf("short application gaps suppressed an active sample: target=%d want 1080000", got)
+	}
+}
+
+func TestAdaptiveMostlyActiveSamplesStillRespondToPath(t *testing.T) {
+	for _, condition := range []string{"slower_delivery", "loss", "rtt"} {
+		t.Run(condition, func(t *testing.T) {
+			controller := newTestAdaptive(t, ProfileBalanced)
+			snapshot := Snapshot{At: time.Unix(1, 0), MinRTT: time.Millisecond, SmoothedRTT: time.Millisecond}
+			mustObserveIdleSnapshot(t, controller, snapshot)
+			snapshot.At = snapshot.At.Add(time.Second)
+			snapshot.SentBytes = 1_000_000
+			mustObserveIdleSnapshot(t, controller, snapshot)
+			learned := controller.TargetBytesPerSecond()
+			for range deliveryRateWindow {
+				// A one-second paced or transport-blocked write is active demand.
+				// Its 20ms source-read gap exceeds the 10ms sample window, but
+				// must not discard the much longer active congestion observation.
+				snapshot.At = snapshot.At.Add(time.Second + 20*time.Millisecond)
+				snapshot.ApplicationIdleTime += 20 * time.Millisecond
+				delivered := uint64(1_000_000)
+				switch condition {
+				case "slower_delivery":
+					delivered = 100_000
+				case "loss":
+					snapshot.LostBytes += 200_000
+				case "rtt":
+					snapshot.SmoothedRTT = 2 * time.Millisecond
+				}
+				snapshot.SentBytes += delivered
+				beforeNext := controller.estimator.next
+				mustObserveIdleSnapshot(t, controller, snapshot)
+				if controller.estimator.next != (beforeNext+1)%deliveryRateWindow {
+					t.Fatal("mostly active interval did not update bandwidth history")
+				}
+			}
+			if got := controller.TargetBytesPerSecond(); got >= learned {
+				t.Fatalf("mostly active %s did not reduce target: %d >= %d", condition, got, learned)
+			}
+		})
+	}
+}
+
+func TestAdaptiveApplicationIdleHalfIntervalBoundary(t *testing.T) {
+	for _, test := range []struct {
+		name     string
+		idle     time.Duration
+		filtered bool
+	}{
+		{name: "just below half", idle: 12*time.Millisecond - time.Nanosecond},
+		{name: "exactly half", idle: 12 * time.Millisecond, filtered: true},
+		{name: "above half", idle: 13 * time.Millisecond, filtered: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			controller := newTestAdaptive(t, ProfileBalanced)
+			base := Snapshot{At: time.Unix(1, 0), MinRTT: time.Millisecond, SmoothedRTT: time.Millisecond}
+			mustObserveIdleSnapshot(t, controller, base)
+			sample := base
+			sample.At = base.At.Add(24 * time.Millisecond)
+			sample.SentBytes = 24_000
+			sample.ApplicationIdleTime = test.idle
+			mustObserveIdleSnapshot(t, controller, sample)
+			wantRate, wantCount := int64(1_080_000), 1
+			if test.filtered {
+				wantRate, wantCount = 1_000_000, 0
+			}
+			if controller.TargetBytesPerSecond() != wantRate || controller.estimator.count != wantCount {
+				t.Fatalf("idle=%s: target/history = %d/%d, want %d/%d", test.idle,
+					controller.TargetBytesPerSecond(), controller.estimator.count, wantRate, wantCount)
+			}
+			if controller.previous != sample {
+				t.Fatal("stable sample did not rebaseline counters")
+			}
+		})
+	}
+}
+
+func TestAdaptiveApplicationIdleRejectsNegativeTime(t *testing.T) {
+	for _, established := range []bool{false, true} {
+		controller := newTestAdaptive(t, ProfileBalanced)
+		base := Snapshot{At: time.Unix(1, 0), MinRTT: time.Millisecond, SmoothedRTT: time.Millisecond}
+		if established {
+			mustObserveIdleSnapshot(t, controller, base)
+		}
+		before := controller.previous
+		invalid := base
+		invalid.At = base.At.Add(time.Second)
+		invalid.ApplicationIdleTime = -time.Nanosecond
+		if err := controller.Observe(invalid); !errors.Is(err, ErrInvalidSnapshot) {
+			t.Fatalf("negative idle time error=%v want ErrInvalidSnapshot", err)
+		}
+		if controller.previous != before || controller.haveSnapshot != established || controller.TargetBytesPerSecond() != 1_000_000 {
+			t.Fatal("invalid idle time changed controller state")
+		}
+	}
+}
+
+func TestAdaptiveApplicationIdleCounterResetStartsNewEpoch(t *testing.T) {
+	for _, reset := range []string{"idle", "sent"} {
+		t.Run(reset, func(t *testing.T) {
+			controller := newTestAdaptive(t, ProfileBalanced)
+			base := Snapshot{
+				At: time.Unix(1, 0), SentBytes: 1000, ApplicationIdleTime: time.Second,
+				MinRTT: time.Millisecond, SmoothedRTT: time.Millisecond,
+			}
+			mustObserveIdleSnapshot(t, controller, base)
+			active := base
+			active.At = active.At.Add(time.Second)
+			active.SentBytes += 2_000_000
+			mustObserveIdleSnapshot(t, controller, active)
+			if controller.TargetBytesPerSecond() == 1_000_000 || controller.estimator.count == 0 {
+				t.Fatal("test did not establish a learned rate")
+			}
+			resetSample := active
+			resetSample.At = resetSample.At.Add(time.Millisecond) // Reset is recognized even below the sample window.
+			if reset == "idle" {
+				resetSample.ApplicationIdleTime = 0
+			} else {
+				resetSample.SentBytes = 0
+			}
+			mustObserveIdleSnapshot(t, controller, resetSample)
+			if controller.TargetBytesPerSecond() != 1_000_000 || controller.estimator.count != 0 || controller.previous != resetSample {
+				t.Fatal("counter reset did not restore the initial rate and rebaseline")
+			}
+		})
+	}
+}
+
+func TestAdaptiveApplicationIdleRecoveryStillRespondsToActivePath(t *testing.T) {
+	for _, condition := range []string{"slower_delivery", "loss", "rtt"} {
+		t.Run(condition, func(t *testing.T) {
+			controller := newTestAdaptive(t, ProfileBalanced)
+			snapshot := Snapshot{At: time.Unix(1, 0), MinRTT: time.Millisecond, SmoothedRTT: time.Millisecond}
+			mustObserveIdleSnapshot(t, controller, snapshot)
+			snapshot.At = snapshot.At.Add(time.Second)
+			snapshot.SentBytes = 1_000_000
+			mustObserveIdleSnapshot(t, controller, snapshot)
+			learned := controller.TargetBytesPerSecond()
+			beforeEstimator := controller.estimator
+			snapshot.At = snapshot.At.Add(2500 * time.Millisecond)
+			snapshot.SentBytes += 20_000
+			snapshot.ApplicationIdleTime = 2500 * time.Millisecond
+			mustObserveIdleSnapshot(t, controller, snapshot)
+			if controller.TargetBytesPerSecond() != learned || controller.estimator != beforeEstimator {
+				t.Fatal("idle window discarded learned capacity")
+			}
+			for range deliveryRateWindow {
+				snapshot.At = snapshot.At.Add(time.Second)
+				delivered := uint64(1_000_000)
+				switch condition {
+				case "slower_delivery":
+					delivered = 100_000
+				case "loss":
+					snapshot.LostBytes += 200_000
+				case "rtt":
+					snapshot.SmoothedRTT = 2 * time.Millisecond
+				}
+				snapshot.SentBytes += delivered
+				mustObserveIdleSnapshot(t, controller, snapshot)
+			}
+			if got := controller.TargetBytesPerSecond(); got >= learned {
+				t.Fatalf("resumed active %s did not reduce target: %d >= %d", condition, got, learned)
+			}
+		})
+	}
+}
+
+func TestApplicationIdleSnapshotsDoNotChangeFixedOrRenoModes(t *testing.T) {
+	for _, mode := range []Mode{ModeFixedRate, ModeReno} {
+		t.Run(mode.String(), func(t *testing.T) {
+			controller, err := New(Config{Mode: mode, FixedRateBytesPerSecond: 123_456, Clock: newFakeClock()})
+			if err != nil {
+				t.Fatal(err)
+			}
+			before := controller.TargetBytesPerSecond()
+			for _, idle := range []time.Duration{time.Second, 0, -time.Second} {
+				// Observe has always been an unconditional no-op outside adaptive
+				// mode, even for otherwise invalid adaptive-only observations.
+				mustObserveIdleSnapshot(t, controller, Snapshot{At: time.Unix(1, 0), ApplicationIdleTime: idle})
+			}
+			if controller.TargetBytesPerSecond() != before || controller.haveSnapshot {
+				t.Fatal("idle snapshots changed a non-adaptive controller")
+			}
+		})
+	}
+}
+
+func mustObserveIdleSnapshot(t *testing.T, controller *Controller, snapshot Snapshot) {
+	t.Helper()
+	if err := controller.Observe(snapshot); err != nil {
+		t.Fatal(err)
+	}
+}