From 99b46b5a91db627226d1ced6440079e65f4b897d Mon Sep 17 00:00:00 2001 From: cppla Date: Tue, 22 Sep 2026 21:37:45 +0800 Subject: [PATCH 1/2] fix: exclude application-idle intervals from adaptive pacing samples --- docs/ACCELERATION.md | 18 ++ docs/ARCHITECTURE.md | 6 + internal/accel/pacer.go | 23 +- internal/accel/pacing_idle_test.go | 202 ++++++++++++++ internal/tunnel/datagram.go | 12 +- internal/tunnel/pacing.go | 69 ++++- internal/tunnel/pacing_activity_test.go | 49 ++++ .../tunnel/pacing_datagram_activity_test.go | 106 +++++++ .../tunnel/pacing_idle_integration_test.go | 188 +++++++++++++ internal/tunnel/pacing_write_activity_test.go | 259 ++++++++++++++++++ internal/tunnel/quic.go | 13 + 11 files changed, 932 insertions(+), 13 deletions(-) create mode 100644 internal/accel/pacing_idle_test.go create mode 100644 internal/tunnel/pacing_activity_test.go create mode 100644 internal/tunnel/pacing_datagram_activity_test.go create mode 100644 internal/tunnel/pacing_idle_integration_test.go create mode 100644 internal/tunnel/pacing_write_activity_test.go diff --git a/docs/ACCELERATION.md b/docs/ACCELERATION.md index fb82a56..6d5762b 100644 --- a/docs/ACCELERATION.md +++ b/docs/ACCELERATION.md @@ -58,6 +58,24 @@ 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 a +sample contains at least one stable sampling window of idle time, 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. 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” diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 83d1d65..6e9b4e5 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -205,6 +205,12 @@ 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 with a stable sampling window of no pending writes are rebaselined +instead of turning ACK-only traffic into a low bandwidth estimate. 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`. diff --git a/internal/accel/pacer.go b/internal/accel/pacer.go index a6c2809..e70c5b7 100644 --- a/internal/accel/pacer.go +++ b/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,13 +276,15 @@ 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 { @@ -282,6 +292,13 @@ func (c *Controller) Observe(snapshot Snapshot) error { } c.previous = snapshot + if idleDelta >= window { + // 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. + return nil + } if sentDelta == 0 || snapshot.MinRTT == 0 || snapshot.SmoothedRTT == 0 { return nil } diff --git a/internal/accel/pacing_idle_test.go b/internal/accel/pacing_idle_test.go new file mode 100644 index 0000000..9297743 --- /dev/null +++ b/internal/accel/pacing_idle_test.go @@ -0,0 +1,202 @@ +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 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) + } +} diff --git a/internal/tunnel/datagram.go b/internal/tunnel/datagram.go index 1a55b82..322f4c9 100644 --- a/internal/tunnel/datagram.go +++ b/internal/tunnel/datagram.go @@ -991,8 +991,18 @@ func sendQUICDatagram( pacer *connectionPacer, queued queuedDatagram, ) error { - frameSize := defaultUDPFrameSize frames := queued.frames + if len(frames) == 0 { + return nil + } + if err := context.Cause(queued.ctx); err != nil { + return err + } + if pacer != nil { + pacer.beginWrite() + defer pacer.endWrite() + } + frameSize := defaultUDPFrameSize for attempt := 0; attempt < 2; attempt++ { var err error for index, frame := range frames { diff --git a/internal/tunnel/pacing.go b/internal/tunnel/pacing.go index f103537..b09ba56 100644 --- a/internal/tunnel/pacing.go +++ b/internal/tunnel/pacing.go @@ -60,6 +60,50 @@ type connectionPacer struct { config PacingConfig controller *accel.Controller fixedRate uint64 + activity pacingWriteActivity +} + +// pacingWriteActivity counts the union of pending application writes across +// streams and datagram batches on a connection. Time spent waiting for pacing +// or QUIC flow control is busy time, not application idle time. Its owner holds +// p.mu. +type pacingWriteActivity struct { + activeWrites int + idleSince time.Time + idle time.Duration +} + +func (a *pacingWriteActivity) begin(at time.Time) { + if a.activeWrites == 0 && !a.idleSince.IsZero() { + a.idle += at.Sub(a.idleSince) + } + a.activeWrites++ +} + +func (a *pacingWriteActivity) end(at time.Time) { + a.activeWrites-- + if a.activeWrites == 0 { + a.idleSince = at + } +} + +func (a *pacingWriteActivity) idleTime(at time.Time) time.Duration { + if a.activeWrites == 0 && !a.idleSince.IsZero() { + return a.idle + at.Sub(a.idleSince) + } + return a.idle +} + +func (p *connectionPacer) beginWrite() { + p.mu.Lock() + p.activity.begin(time.Now()) + p.mu.Unlock() +} + +func (p *connectionPacer) endWrite() { + p.mu.Lock() + p.activity.end(time.Now()) + p.mu.Unlock() } func newConnectionPacer(config PacingConfig, fixedRate uint64) (*connectionPacer, error) { @@ -119,18 +163,25 @@ func (p *connectionPacer) setFixedRate(rate uint64) error { func (p *connectionPacer) wait(ctx context.Context, bytes int, conn *quic.Conn) error { p.mu.Lock() controller := p.controller - p.mu.Unlock() if controller == nil { + p.mu.Unlock() return nil } - stats := conn.ConnectionStats() - _ = controller.Observe(accel.Snapshot{ - At: time.Now(), - SentBytes: stats.BytesSent, - LostBytes: stats.BytesLost, - MinRTT: stats.MinRTT, - SmoothedRTT: stats.SmoothedRTT, - }) + if controller.Mode() == accel.ModeAdaptive { + // Keep counter reads and observations ordered across concurrent streams. + // Only the sample is serialized: never hold p.mu during admission or I/O. + stats := conn.ConnectionStats() + now := time.Now() + _ = controller.Observe(accel.Snapshot{ + At: now, + SentBytes: stats.BytesSent, + LostBytes: stats.BytesLost, + MinRTT: stats.MinRTT, + SmoothedRTT: stats.SmoothedRTT, + ApplicationIdleTime: p.activity.idleTime(now), + }) + } + p.mu.Unlock() return controller.Wait(ctx, bytes) } diff --git a/internal/tunnel/pacing_activity_test.go b/internal/tunnel/pacing_activity_test.go new file mode 100644 index 0000000..c53459a --- /dev/null +++ b/internal/tunnel/pacing_activity_test.go @@ -0,0 +1,49 @@ +package tunnel + +import ( + "testing" + "time" +) + +func TestPacingWriteActivityTracksConnectionIdleUnion(t *testing.T) { + var activity pacingWriteActivity + base := time.Unix(1, 0) + check := func(at time.Duration, active int, idle time.Duration) { + t.Helper() + if activity.activeWrites != active { + t.Fatalf("at %s: active = %d, want %d", at, activity.activeWrites, active) + } + if got := activity.idleTime(base.Add(at)); got != idle { + t.Fatalf("at %s: idle = %s, want %s", at, got, idle) + } + } + check(0, 0, 0) // A zero-value tracker has no pre-connection idle interval. + activity.begin(base) + check(time.Second, 1, 0) + activity.begin(base.Add(time.Second)) + activity.end(base.Add(2 * time.Second)) + check(3*time.Second, 1, 0) // A sibling writer still has pending data. + activity.end(base.Add(3 * time.Second)) + check(4*time.Second, 0, time.Second) + check(5*time.Second, 0, 2*time.Second) // Sampling must not double-count. + activity.begin(base.Add(5 * time.Second)) + check(8*time.Second, 1, 2*time.Second) + activity.end(base.Add(9 * time.Second)) + activity.begin(base.Add(12 * time.Second)) + check(15*time.Second, 1, 5*time.Second) +} + +func TestPacingWriteActivityAccumulatesShortSourceGaps(t *testing.T) { + var activity pacingWriteActivity + base := time.Unix(1, 0) + for index := range 20 { + started := base.Add(time.Duration(index) * time.Millisecond) + activity.begin(started) + activity.end(started.Add(400 * time.Microsecond)) + } + // The cumulative counter retains gaps even when an observer's stable + // sampling window spans many short writes. + if got := activity.idleTime(base.Add(20 * time.Millisecond)); got != 12*time.Millisecond { + t.Fatalf("cumulative idle = %s, want 12ms", got) + } +} diff --git a/internal/tunnel/pacing_datagram_activity_test.go b/internal/tunnel/pacing_datagram_activity_test.go new file mode 100644 index 0000000..8c33e68 --- /dev/null +++ b/internal/tunnel/pacing_datagram_activity_test.go @@ -0,0 +1,106 @@ +package tunnel + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/cppla/autocar/internal/accel" +) + +func TestQUICDatagramActivityCoversPacingWaitAndCancellation(t *testing.T) { + clock := &datagramActivityClock{started: make(chan struct{})} + controller, err := accel.New(accel.Config{ + Mode: accel.ModeFixedRate, FixedRateBytesPerSecond: 1, BurstBytes: 1, Clock: clock, + }) + if err != nil { + t.Fatal(err) + } + // Consume the initial burst. The next admission must enter the controlled + // sleep and cannot reach SendDatagram before cancellation. + if err := controller.Wait(context.Background(), 1); err != nil { + t.Fatal(err) + } + pacer := &connectionPacer{controller: controller} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- sendQUICDatagram(nil, pacer, queuedDatagram{ctx: ctx, frames: [][]byte{{1}, {2}}}) + }() + select { + case <-clock.started: + case <-time.After(2 * time.Second): + t.Fatal("datagram pacing wait did not start") + } + pacer.mu.Lock() + active := pacer.activity.activeWrites + before := pacer.activity.idleTime(time.Now()) + after := pacer.activity.idleTime(time.Now().Add(time.Second)) + pacer.mu.Unlock() + if active != 1 || before != after { + t.Fatalf("blocked datagram activity = %d, idle = %s -> %s; want one active sender and no idle accumulation", active, before, after) + } + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("sendQUICDatagram error = %v, want context.Canceled", err) + } + case <-time.After(2 * time.Second): + t.Fatal("cancellation did not end datagram pacing wait") + } + pacer.mu.Lock() + active = pacer.activity.activeWrites + idleSince := pacer.activity.idleSince + idleDelta := pacer.activity.idleTime(idleSince.Add(time.Second)) - pacer.activity.idle + pacer.mu.Unlock() + if active != 0 || idleSince.IsZero() || idleDelta != time.Second { + t.Fatalf("finished datagram activity = %d, idleSince=%s, idleDelta=%s; want resumed idle accounting", active, idleSince, idleDelta) + } +} + +func TestQUICDatagramActivitySkipsCanceledAndEmptyBatches(t *testing.T) { + cause := errors.New("canceled before send") + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(cause) + for _, test := range []struct { + name string + queued queuedDatagram + want error + }{ + {name: "pre-canceled", queued: queuedDatagram{ctx: ctx, frames: [][]byte{{1}}}, want: cause}, + {name: "empty", queued: queuedDatagram{ctx: context.Background()}}, + {name: "empty pre-canceled", queued: queuedDatagram{ctx: ctx}}, + } { + t.Run(test.name, func(t *testing.T) { + // Seed an existing idle epoch: even a balanced begin/end would mutate + // these timestamps and must be detected, not just an active-count leak. + initial := pacingWriteActivity{idleSince: time.Unix(1, 0), idle: time.Millisecond} + pacer := &connectionPacer{activity: initial} + if err := sendQUICDatagram(nil, pacer, test.queued); !errors.Is(err, test.want) { + t.Fatalf("sendQUICDatagram error = %v, want %v", err, test.want) + } + if pacer.activity != initial { + t.Fatalf("skipped batch changed activity: got %+v, want %+v", pacer.activity, initial) + } + }) + } +} + +// A fixed clock prevents real-time token refills. Sleep is cancellation-aware +// and announces admission without polling timing-dependent controller state. +type datagramActivityClock struct { + started chan struct{} + once sync.Once +} + +func (*datagramActivityClock) Now() time.Time { return time.Unix(1, 0) } + +func (c *datagramActivityClock) Sleep(ctx context.Context, _ time.Duration) error { + c.once.Do(func() { close(c.started) }) + <-ctx.Done() + return context.Cause(ctx) +} diff --git a/internal/tunnel/pacing_idle_integration_test.go b/internal/tunnel/pacing_idle_integration_test.go new file mode 100644 index 0000000..3a389db --- /dev/null +++ b/internal/tunnel/pacing_idle_integration_test.go @@ -0,0 +1,188 @@ +package tunnel + +import ( + "bytes" + "context" + "fmt" + "io" + "testing" + "time" + + quic "github.com/quic-go/quic-go" +) + +// This exercises the production QUIC stream adapter and controller against +// real connection statistics. Authentication and serverCore are intentionally +// outside this focused regression; the peer is a controlled loopback fixture. +func TestQUICPacingReceiveOnlyIntervalDoesNotCollapseSender(t *testing.T) { + const ( + minimumRate = 64 << 10 + initialRate = 8_000_000 + chunkSize = 1024 + chunks = 8 + ) + upload := bytes.Repeat([]byte("receive-only-gap"), chunkSize*chunks/16+1)[:chunkSize*chunks] + download := bytes.Repeat([]byte("same-QUIC-connection"), 1024) + trailer := []byte("upload still works after the receive-only interval") + ready := []byte("ready") + serverTLS, clientTLS := testTLSConfigs(t) + listener, err := quic.ListenAddr("127.0.0.1:0", mustServerTLSConfig(t, serverTLS), hardenedQUICServerConfig(nil, 1)) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + deadline, _ := ctx.Deadline() + type result struct { + before, after int64 + sentDuringGap uint64 + err error + } + serverDone := make(chan result, 1) + go func() { + var observation result + defer func() { serverDone <- observation }() + conn, err := listener.Accept(ctx) + if err != nil { + observation.err = err + return + } + defer conn.CloseWithError(applicationShutdown, "test finished") + stream, err := conn.AcceptStream(ctx) + if err != nil { + observation.err = err + return + } + pacer, err := newConnectionPacer(PacingConfig{ + InitialRateBytesPerSecond: initialRate, + MinRateBytesPerSecond: minimumRate, + }, 0) + if err != nil { + observation.err = err + return + } + wrapped := newQUICStreamConn(stream, conn, pacer) + defer wrapped.Close() + if err := wrapped.SetDeadline(deadline); err != nil { + observation.err = err + return + } + checkRead := func(want []byte) error { + got := make([]byte, len(want)) + if _, err := io.ReadFull(wrapped, got); err != nil { + return err + } + if !bytes.Equal(got, want) { + return fmt.Errorf("server received incorrect %d-byte payload", len(want)) + } + return nil + } + if err := checkRead([]byte{1}); err != nil { + observation.err = err + return + } + if _, err := wrapped.Write(ready); err != nil { + observation.err = err + return + } + observation.before = pacer.controller.TargetBytesPerSecond() + sentBefore := conn.ConnectionStats().BytesSent + // Only the receive direction carries application data during this gap. + // QUIC still emits ACK/control packets, which are not delivery-rate + // evidence that this application's otherwise idle sender is slow. + if err := checkRead(upload); err != nil { + observation.err = err + return + } + observation.sentDuringGap = conn.ConnectionStats().BytesSent - sentBefore + if _, err := wrapped.Write(download); err != nil { + observation.err = err + return + } + observation.after = pacer.controller.TargetBytesPerSecond() + observation.err = checkRead(trailer) + }() + var client *quic.Conn + joined := false + t.Cleanup(func() { + cancel() + if client != nil { + _ = client.CloseWithError(applicationShutdown, "test finished") + } + _ = listener.Close() + if !joined { + select { + case <-serverDone: + case <-time.After(time.Second): + t.Error("QUIC pacing fixture did not stop") + } + } + }) + rawClientTLS, err := clientTLSConfig(clientTLS, listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + client, err = quic.DialAddr(ctx, listener.Addr().String(), rawClientTLS, hardenedQUICClientConfig(nil)) + if err != nil { + t.Fatal(err) + } + stream, err := client.OpenStreamSync(ctx) + if err != nil { + t.Fatal(err) + } + if err := stream.SetDeadline(deadline); err != nil { + t.Fatal(err) + } + if _, err := stream.Write([]byte{1}); err != nil { + t.Fatal(err) + } + gotReady := make([]byte, len(ready)) + if _, err := io.ReadFull(stream, gotReady); err != nil { + t.Fatal(err) + } + if !bytes.Equal(gotReady, ready) { + t.Fatalf("greeting = %q", gotReady) + } + ticker := time.NewTicker(40 * time.Millisecond) + defer ticker.Stop() + for offset := 0; offset < len(upload); offset += chunkSize { + select { + case <-ticker.C: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + if _, err := stream.Write(upload[offset : offset+chunkSize]); err != nil { + t.Fatal(err) + } + } + gotDownload := make([]byte, len(download)) + if _, err := io.ReadFull(stream, gotDownload); err != nil { + t.Fatal(err) + } + if !bytes.Equal(gotDownload, download) { + t.Fatal("download payload mismatch") + } + if _, err := stream.Write(trailer); err != nil { + t.Fatal(err) + } + select { + case observation := <-serverDone: + joined = true + if observation.err != nil { + t.Fatal(observation.err) + } + if observation.sentDuringGap == 0 { + t.Fatal("fixture did not produce outgoing QUIC control traffic during upload") + } + if observation.before != initialRate { + t.Fatalf("initial server target = %d, want %d", observation.before, initialRate) + } + if observation.after != observation.before { + t.Fatalf("receive-only interval changed server target from %d to %d (floor %d, %d control bytes)", + observation.before, observation.after, minimumRate, observation.sentDuringGap) + } + t.Logf("same-connection server target %d -> %d; %d outgoing QUIC control bytes during receive-only interval", + observation.before, observation.after, observation.sentDuringGap) + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } +} diff --git a/internal/tunnel/pacing_write_activity_test.go b/internal/tunnel/pacing_write_activity_test.go new file mode 100644 index 0000000..37badff --- /dev/null +++ b/internal/tunnel/pacing_write_activity_test.go @@ -0,0 +1,259 @@ +package tunnel + +import ( + "context" + "errors" + "io" + "net" + "sync/atomic" + "testing" + "time" + + quic "github.com/quic-go/quic-go" +) + +func TestQUICWriteActivityCoversPacingAndTransportWait(t *testing.T) { + stream := &blockingQUICStream{ + writeStarted: make(chan struct{}), writeCanceled: make(chan struct{}), + } + waitStarted := make(chan struct{}) + allowWrite := make(chan struct{}) + pacer := &activityQUICWritePacer{waitFn: func(ctx context.Context) error { + close(waitStarted) + select { + case <-allowWrite: + return nil + case <-ctx.Done(): + return context.Cause(ctx) + } + }} + conn := newQUICStreamConnWithPacer(stream, nil, pacer) + defer conn.Close() + done := make(chan error, 1) + go func() { _, err := conn.Write([]byte("pending")); done <- err }() + awaitActivitySignal(t, waitStarted) + assertWriteActivity(t, pacer, 1, 1, 0) + close(allowWrite) + awaitActivitySignal(t, stream.writeStarted) + assertWriteActivity(t, pacer, 1, 1, 0) + _ = conn.Close() + if err := awaitActivityResult(t, done); !errors.Is(err, net.ErrClosed) { + t.Fatalf("Write error = %v, want net.ErrClosed", err) + } + assertWriteActivity(t, pacer, 0, 1, 1) +} + +func TestQUICWriteActivityEndsOnEveryReturn(t *testing.T) { + sentinel := errors.New("write failure") + for _, test := range []struct { + name string + writeN int + writeErr error + waitErr error + wantN int + wantErr error + wantWrites int + }{ + {name: "chunked success", writeN: -1, wantN: 7, wantWrites: 3}, + {name: "transport error", writeErr: sentinel, wantErr: sentinel, wantWrites: 1}, + {name: "partial transport error", writeN: 1, writeErr: sentinel, wantN: 1, wantErr: sentinel, wantWrites: 1}, + {name: "short write", writeN: 1, wantN: 1, wantErr: io.ErrShortWrite, wantWrites: 1}, + {name: "pacing error", waitErr: sentinel, wantErr: sentinel}, + } { + t.Run(test.name, func(t *testing.T) { + pacer := &activityQUICWritePacer{maxChunk: 3} + pacer.waitFn = func(context.Context) error { + assertWriteActivity(t, pacer, 1, 1, 0) + return test.waitErr + } + writes := 0 + stream := &activityQUICStream{writeFn: func(p []byte) (int, error) { + assertWriteActivity(t, pacer, 1, 1, 0) + writes++ + if test.writeN < 0 { + return len(p), test.writeErr + } + return test.writeN, test.writeErr + }} + conn := newQUICStreamConnWithPacer(stream, nil, pacer) + defer conn.Close() + n, err := conn.Write([]byte("payload")) + if n != test.wantN || !errors.Is(err, test.wantErr) { + t.Fatalf("Write = %d, %v; want %d, %v", n, err, test.wantN, test.wantErr) + } + if writes != test.wantWrites { + t.Fatalf("transport writes = %d, want %d", writes, test.wantWrites) + } + assertWriteActivity(t, pacer, 0, 1, 1) + }) + } +} + +func TestQUICWriteActivityEndsWhenPacingCanceled(t *testing.T) { + for _, cancellation := range []string{"local close", "peer cancel", "deadline"} { + t.Run(cancellation, func(t *testing.T) { + streamCtx, cancelStream := context.WithCancel(context.Background()) + defer cancelStream() + stream := &blockingQUICStream{ + ctx: streamCtx, writeStarted: make(chan struct{}), writeCanceled: make(chan struct{}), + } + started := make(chan struct{}) + pacer := &activityQUICWritePacer{waitFn: func(ctx context.Context) error { + select { + case <-started: + default: + close(started) + } + <-ctx.Done() + return context.Cause(ctx) + }} + conn := newQUICStreamConnWithPacer(stream, nil, pacer) + defer conn.Close() + done := make(chan error, 1) + go func() { _, err := conn.Write([]byte("pending")); done <- err }() + awaitActivitySignal(t, started) + assertWriteActivity(t, pacer, 1, 1, 0) + wantErr := net.ErrClosed + switch cancellation { + case "local close": + _ = conn.Close() + case "peer cancel": + cancelStream() + case "deadline": + _ = conn.SetWriteDeadline(time.Now()) + wantErr = context.DeadlineExceeded + } + if err := awaitActivityResult(t, done); !errors.Is(err, wantErr) { + t.Fatalf("Write error = %v, want %v", err, wantErr) + } + assertWriteActivity(t, pacer, 0, 1, 1) + select { + case <-stream.writeStarted: + t.Fatal("canceled pacing wait reached the transport") + default: + } + }) + } +} + +func TestQUICWriteActivitySerialWrites(t *testing.T) { + pacer := &activityQUICWritePacer{maxChunk: 2} + conn := newQUICStreamConnWithPacer(&recordingQUICStream{}, nil, pacer) + defer conn.Close() + for i := int64(1); i <= 3; i++ { + pacer.waitFn = func(context.Context) error { + assertWriteActivity(t, pacer, 1, i, i-1) + return nil + } + if _, err := conn.Write([]byte("payload")); err != nil { + t.Fatal(err) + } + assertWriteActivity(t, pacer, 0, i, i) + } +} + +func TestQUICWriteActivitySharedAcrossStreams(t *testing.T) { + started := make(chan struct{}, 2) + pacer := &activityQUICWritePacer{waitFn: func(ctx context.Context) error { + started <- struct{}{} + <-ctx.Done() + return context.Cause(ctx) + }} + first := newQUICStreamConnWithPacer(&recordingQUICStream{}, nil, pacer) + second := newQUICStreamConnWithPacer(&recordingQUICStream{}, nil, pacer) + defer first.Close() + defer second.Close() + firstDone, secondDone := make(chan error, 1), make(chan error, 1) + go func() { _, err := first.Write([]byte("one")); firstDone <- err }() + go func() { _, err := second.Write([]byte("two")); secondDone <- err }() + awaitActivitySignal(t, started) + awaitActivitySignal(t, started) + assertWriteActivity(t, pacer, 2, 2, 0) + _ = first.Close() + if err := awaitActivityResult(t, firstDone); !errors.Is(err, net.ErrClosed) { + t.Fatalf("first Write error = %v", err) + } + assertWriteActivity(t, pacer, 1, 2, 1) + _ = second.Close() + if err := awaitActivityResult(t, secondDone); !errors.Is(err, net.ErrClosed) { + t.Fatalf("second Write error = %v", err) + } + assertWriteActivity(t, pacer, 0, 2, 2) +} + +func TestQUICWriteActivitySkipsZeroAndClosedWrites(t *testing.T) { + for _, closing := range []string{"open", "close", "close write"} { + t.Run(closing, func(t *testing.T) { + pacer := &activityQUICWritePacer{} + stream := &recordingQUICStream{} + conn := newQUICStreamConnWithPacer(stream, nil, pacer) + defer conn.Close() + switch closing { + case "close": + _ = conn.Close() + case "close write": + _ = conn.CloseWrite() + } + n, err := conn.Write(nil) + if n != 0 || (closing == "open" && err != nil) || (closing != "open" && !errors.Is(err, net.ErrClosed)) { + t.Fatalf("zero Write = %d, %v", n, err) + } + if closing != "open" { + if n, err := conn.Write([]byte("closed")); n != 0 || !errors.Is(err, net.ErrClosed) { + t.Fatalf("closed Write = %d, %v", n, err) + } + } + assertWriteActivity(t, pacer, 0, 0, 0) + }) + } +} + +type activityQUICWritePacer struct { + active, begins, ends atomic.Int64 + maxChunk int + waitFn func(context.Context) error +} + +func (p *activityQUICWritePacer) beginWrite() { p.active.Add(1); p.begins.Add(1) } +func (p *activityQUICWritePacer) endWrite() { p.active.Add(-1); p.ends.Add(1) } +func (p *activityQUICWritePacer) maxChunkBytes() int { return p.maxChunk } +func (p *activityQUICWritePacer) wait(ctx context.Context, _ int, _ *quic.Conn) error { + if p.waitFn != nil { + return p.waitFn(ctx) + } + return nil +} + +type activityQUICStream struct { + recordingQUICStream + writeFn func([]byte) (int, error) +} + +func (s *activityQUICStream) Write(p []byte) (int, error) { return s.writeFn(p) } + +func assertWriteActivity(t *testing.T, p *activityQUICWritePacer, active, begins, ends int64) { + t.Helper() + if gotActive, gotBegins, gotEnds := p.active.Load(), p.begins.Load(), p.ends.Load(); gotActive != active || gotBegins != begins || gotEnds != ends { + t.Fatalf("write activity = (%d active, %d begins, %d ends), want (%d, %d, %d)", gotActive, gotBegins, gotEnds, active, begins, ends) + } +} + +func awaitActivitySignal(t *testing.T, signal <-chan struct{}) { + t.Helper() + select { + case <-signal: + case <-time.After(2 * time.Second): + t.Fatal("write activity signal did not arrive") + } +} + +func awaitActivityResult(t *testing.T, done <-chan error) error { + t.Helper() + select { + case err := <-done: + return err + case <-time.After(2 * time.Second): + t.Fatal("write activity did not finish") + return nil + } +} diff --git a/internal/tunnel/quic.go b/internal/tunnel/quic.go index 9636aeb..8bab686 100644 --- a/internal/tunnel/quic.go +++ b/internal/tunnel/quic.go @@ -1155,6 +1155,13 @@ type quicWritePacer interface { maxChunkBytes() int } +// quicWriteActivity optionally tracks application write demand across pacing +// waits and transport backpressure, rather than just time spent admitting bytes. +type quicWriteActivity interface { + beginWrite() + endWrite() +} + type quicStreamConn struct { stream quicStream conn *quic.Conn @@ -1208,6 +1215,12 @@ func (c *quicStreamConn) Write(p []byte) (int, error) { if closed { return 0, net.ErrClosed } + if len(p) > 0 { + if activity, ok := c.pacer.(quicWriteActivity); ok { + activity.beginWrite() + defer activity.endWrite() + } + } chunkSize := len(p) if c.pacer != nil { maxChunk := c.pacer.maxChunkBytes() From 0d7637ab6035c9489e18e9961fef40a4b02c7ab7 Mon Sep 17 00:00:00 2001 From: cppla Date: Tue, 22 Sep 2026 21:41:00 +0800 Subject: [PATCH 2/2] fix: retain predominantly busy pacing observations --- docs/ACCELERATION.md | 12 +++-- docs/ARCHITECTURE.md | 5 +- internal/accel/pacer.go | 4 +- internal/accel/pacing_idle_test.go | 73 ++++++++++++++++++++++++++++++ 4 files changed, 86 insertions(+), 8 deletions(-) diff --git a/docs/ACCELERATION.md b/docs/ACCELERATION.md index 6d5762b..31127c6 100644 --- a/docs/ACCELERATION.md +++ b/docs/ACCELERATION.md @@ -61,8 +61,8 @@ 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 a -sample contains at least one stable sampling window of idle time, the controller +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 @@ -72,9 +72,11 @@ 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. Subsequent active samples can still reduce the -target when capacity, RTT or loss changes. Fixed-rate and bypass modes are -unchanged. +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 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 6e9b4e5..359372f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -206,8 +206,9 @@ 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 with a stable sampling window of no pending writes are rebaselined -instead of turning ACK-only traffic into a low bandwidth estimate. Pending +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. diff --git a/internal/accel/pacer.go b/internal/accel/pacer.go index e70c5b7..323b480 100644 --- a/internal/accel/pacer.go +++ b/internal/accel/pacer.go @@ -292,11 +292,13 @@ func (c *Controller) Observe(snapshot Snapshot) error { } c.previous = snapshot - if idleDelta >= window { + 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 { diff --git a/internal/accel/pacing_idle_test.go b/internal/accel/pacing_idle_test.go index 9297743..6a4ac50 100644 --- a/internal/accel/pacing_idle_test.go +++ b/internal/accel/pacing_idle_test.go @@ -84,6 +84,79 @@ func TestAdaptiveApplicationShortIdleGapsStillUpdateRate(t *testing.T) { } } +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)