diff --git a/docs/ACCELERATION.md b/docs/ACCELERATION.md index fb82a56..31127c6 100644 --- a/docs/ACCELERATION.md +++ b/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” diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 83d1d65..359372f 100644 --- a/docs/ARCHITECTURE.md +++ b/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`. diff --git a/internal/accel/pacer.go b/internal/accel/pacer.go index a6c2809..323b480 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,15 @@ func (c *Controller) Observe(snapshot Snapshot) error { } 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 } diff --git a/internal/accel/pacing_idle_test.go b/internal/accel/pacing_idle_test.go new file mode 100644 index 0000000..6a4ac50 --- /dev/null +++ b/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) + } +} 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()